1#!/usr/bin/env python3
2import importlib
3import json
4import os
5from pathlib import Path
6import sqlite3
7import tempfile
8import unittest
9from unittest.mock import patch
10
11shale = importlib.import_module('shale-labels')
12
13
14class ShaleTests(unittest.TestCase):
15 def setUp(self):
16 self.temp = tempfile.TemporaryDirectory()
17 self.root = Path(self.temp.name)
18 self.env = patch.dict(os.environ, {'STUDIO_SHALE_DATABASE': str(self.root / 'shale.sqlite'),
19 'STUDIO_DASHBOARD_DATA': str(self.root), 'STUDIO_DOMAIN': 'studio.test', 'STUDIO_SHALE_POLICY': str(self.root / 'policy.json')})
20 self.env.start()
21 (self.root / 'policy.json').write_text(json.dumps({'actorOwners': {'ai': 'owner'}, 'statusDelegates': {'ai': ['chat']}}))
22 with sqlite3.connect(self.root / 'accounts.sqlite') as db:
23 db.execute("CREATE TABLE users(id TEXT PRIMARY KEY,profile TEXT,username TEXT GENERATED ALWAYS AS (json_extract(profile,'$.username')) STORED UNIQUE)")
24 for name in ('owner', 'other', 'ai'):
25 db.execute('INSERT INTO users(id,profile) VALUES(?,?)', (name, json.dumps({'username': name, 'enabled': True})))
26 self.grant = {'id': 'fixture', 'user': 'owner', 'resource': 'https://snowglobe.studio.test/mcp/shale', 'scopes': ['shale:write'], 'resources': ['chat']}
27 with sqlite3.connect(self.root / 'connections.sqlite') as db:
28 db.execute('CREATE TABLE records(key TEXT PRIMARY KEY,value TEXT,last_updated INTEGER)')
29 db.execute('INSERT INTO records VALUES(?,?,0)', ('grant:fixture', json.dumps(self.grant)))
30 for name in ('owner', 'other', 'ai'):
31 db.execute('INSERT INTO records VALUES(?,?,0)', ('shale-session:' + name, json.dumps({'origin': 'https://shale.studio.test/'})))
32 self.db = sqlite3.connect(self.root / 'shale.sqlite')
33 self.db.executescript('''
34 CREATE TABLE users(id INTEGER PRIMARY KEY,uuid TEXT,name TEXT,provider TEXT,snowflake TEXT,joined_on TEXT,last_updated TEXT);
35 CREATE TABLE repositories(id INTEGER PRIMARY KEY,name TEXT,owner INTEGER,last_updated TEXT,open_issues INTEGER DEFAULT 0,access_issues TEXT DEFAULT 'public',access_issues_submit TEXT DEFAULT 'public',access_issues_comment TEXT DEFAULT 'public');
36 CREATE TABLE issues(id INTEGER PRIMARY KEY,uuid TEXT,repo INTEGER,number INTEGER,status TEXT,last_updated TEXT,author INTEGER DEFAULT 3,title TEXT DEFAULT 'Title');
37 CREATE TABLE issue_labels(id INTEGER PRIMARY KEY,uuid TEXT,repo INTEGER REFERENCES repositories(id),name TEXT,description TEXT,color TEXT,open_issues INTEGER DEFAULT 0,last_updated TEXT);
38 CREATE TABLE issues__labels(id INTEGER PRIMARY KEY,uuid TEXT,issue INTEGER REFERENCES issues(id),label INTEGER REFERENCES issue_labels(id));
39 CREATE TABLE issue_actions(id INTEGER PRIMARY KEY,uuid TEXT,issue INTEGER REFERENCES issues(id),actor INTEGER REFERENCES users(id),kind TEXT,payload TEXT,last_updated TEXT,added_on TEXT,deleted_on TEXT);
40 INSERT INTO users(id,name,provider,snowflake) VALUES(1,'owner','snowglobe.studio.test','owner'),(2,'other','snowglobe.studio.test','other'),(3,'ai','snowglobe.studio.test','ai');
41 INSERT INTO repositories(id,name,owner,last_updated) VALUES(1,'chat',1,''),(2,'other',2,'');
42 INSERT INTO issues(id,repo,number,status,last_updated) VALUES(1,1,1,'todo',''),(2,1,2,'done',''),(3,2,1,'todo','');
43 ''')
44 self.db.commit()
45
46 def tearDown(self):
47 self.db.close()
48 self.env.stop()
49 self.temp.cleanup()
50
51 def call(self, action, **args):
52 return shale.handle({'action': action, 'arguments': {'model': 'gpt-6.1-sol', **args}, 'grant': 'fixture', 'repository': 'chat'})
53
54 def grant_as(self, **change):
55 with sqlite3.connect(self.root / 'connections.sqlite') as db:
56 db.execute('UPDATE records SET value=? WHERE key=?', (json.dumps({**self.grant, **change}), 'grant:fixture'))
57
58 def create_label(self, name='bug'):
59 return self.call('create_label', name=name, description='Description', color='#123456')['label']['id']
60
61 def test_lifecycle_counts_history_and_protected_deletion(self):
62 a, b = self.create_label(), self.create_label('feature')
63 self.call('set_issue_labels', id=1, labels=[a, b])
64 self.call('set_issue_labels', id=2, labels=[a])
65 self.call('set_issue_labels', id=1, labels=[a, b])
66 self.assertEqual(self.db.execute('SELECT count(*) FROM issue_actions').fetchone(), (3,))
67 self.assertEqual(self.call('update_label', label_id=a, name='defect', description='', color='#abcdef')['label']['openIssues'], 1)
68 with self.assertRaises(shale.Error):
69 self.call('delete_label', label_id=a)
70 self.call('set_issue_labels', id=1, labels=[b])
71 self.call('set_issue_labels', id=2, labels=[])
72 with self.assertRaises(shale.Error):
73 self.call('delete_label', label_id=a)
74 unused = self.create_label('unused')
75 self.call('delete_label', label_id=unused)
76 self.assertEqual(self.db.execute('SELECT label FROM issues__labels').fetchall(), [(b,)])
77 self.assertEqual(self.db.execute('SELECT DISTINCT u.name FROM issue_actions a JOIN users u ON a.actor=u.id').fetchall(), [('ai-owner-gpt-6.1-sol',)])
78 self.assertEqual(self.db.execute('PRAGMA foreign_key_check').fetchall(), [])
79
80 def test_models_and_owners_are_distinct_durable_oauth_subjects(self):
81 created = self.call('create_issue', title='Question', description='Body', status='needs_triage')
82 number = created['id']
83 self.call('comment_issue', id=number, comment='Claude', model='claude-opus-5.5')
84 self.call('set_issue_title', id=number, title='Updated')
85 self.grant_as(user='other')
86 self.call('comment_issue', id=number, comment='Other owner')
87 self.assertEqual(self.db.execute('SELECT u.name FROM issue_actions a JOIN users u ON a.actor=u.id ORDER BY a.id').fetchall(),
88 [('ai-owner-gpt-6.1-sol',), ('ai-owner-claude-opus-5.5',), ('ai-owner-gpt-6.1-sol',), ('ai-other-gpt-6.1-sol',)])
89 with self.assertRaises(shale.Error):
90 self.call('set_issue_title', id=number, title='Cannot take ownership')
91 self.assertEqual(self.db.execute('SELECT status,title FROM issues WHERE number=? AND repo=1', (number,)).fetchone(), ('needs_triage', 'Updated'))
92 with sqlite3.connect(self.root / 'accounts.sqlite') as db:
93 rows = db.execute("SELECT id,profile FROM users WHERE json_extract(profile,'$.kind')='ai'").fetchall()
94 self.assertEqual(len(rows), 3)
95 self.assertEqual(len({uid for uid, _ in rows}), 3)
96 self.assertTrue(all(json.loads(profile)['owner'] in ('owner', 'other') for _, profile in rows))
97
98 def test_explicit_legacy_delegation_keeps_owner_model_and_status_counts(self):
99 label = self.create_label()
100 self.call('set_issue_labels', id=1, labels=[label])
101 self.call('set_issue_status', id=1, status='done')
102 self.grant_as(user='ai')
103 self.call('set_issue_status', id=1, status='todo', model='claude-opus-5.5')
104 self.call('set_issue_status', id=1, status='todo', model='claude-opus-5.5')
105 self.assertEqual(self.db.execute("SELECT u.name,a.payload FROM issue_actions a JOIN users u ON a.actor=u.id WHERE kind='update_status' ORDER BY a.id").fetchall(), [('ai-owner-gpt-6.1-sol', 'done'), ('ai-owner-claude-opus-5.5', 'todo')])
106 self.assertEqual(self.db.execute('SELECT open_issues FROM issue_labels WHERE id=?', (label,)).fetchone(), (1,))
107 self.db.execute('UPDATE repositories SET open_issues=99 WHERE id=1'); self.db.commit()
108 self.assertEqual(shale.metadata({'repository': 'chat', 'accounts': []})['openIssues'], 1)
109 with self.assertRaises(shale.Error):
110 self.create_label('owner-only')
111 (self.root / 'policy.json').write_text('{}')
112 with self.assertRaises(shale.Error):
113 self.call('set_issue_status', id=1, status='done')
114
115 def test_metadata_never_reads_trailers_and_only_returns_requested_accounts(self):
116 result = self.call('create_issue', title='Task', description='Assisted-by: wrong-model', status='todo')
117 name = result['actor'][1:]
118 value = shale.metadata({'repository': 'chat', 'accounts': [name, 'owner', 'ai']})
119 self.assertEqual(value['accounts'], {name: {'model': 'gpt-6.1-sol', 'owner': 'owner', 'picture': None}})
120 subject = self.db.execute('SELECT snowflake FROM users WHERE name=?', (name,)).fetchone()[0]
121 self.db.execute("INSERT INTO users(name,provider,snowflake) VALUES('foreign-owner','foreign.test','owner'),('ai-foreign-human','foreign.test',?)", (subject,))
122 self.db.commit()
123 self.assertEqual(shale.metadata({'repository': 'chat', 'accounts': ['ai-foreign-human']})['accounts'], {})
124 self.assertEqual(shale.metadata({'repository': None, 'accounts': []}), {'accounts': {}, 'openIssues': None})
125 for names in ([True], ['x'] * 1025):
126 with self.assertRaises(shale.Error):
127 shale.metadata({'repository': 'chat', 'accounts': names})
128
129 def test_native_avatars_use_the_same_owner_picture_as_model_credits(self):
130 name = self.call('create_issue', title='Task', description='', status='todo')['actor'][1:]
131 with sqlite3.connect(self.root / 'accounts.sqlite') as db:
132 row = db.execute("SELECT profile FROM users WHERE id='owner'").fetchone()
133 profile = json.loads(row[0]); profile['attributes'] = {'picture': ['https://snowglobe.test/picture']}
134 db.execute("UPDATE users SET profile=? WHERE id='owner'", (json.dumps(profile),))
135 self.db.execute("UPDATE users SET uuid='native-owner' WHERE id=1"); self.db.commit()
136 metadata = shale.metadata({'repository': 'chat', 'accounts': [name, 'owner']})['accounts']
137 self.assertEqual(metadata['owner'], {'avatar': 'native-owner', 'picture': 'https://snowglobe.test/picture'})
138 self.assertEqual(metadata[name]['picture'], metadata['owner']['picture'])
139 self.assertEqual(shale.metadata({'repository': None, 'accounts': []})['accounts'], {})
140
141 def test_scope_revocation_and_private_permissions(self):
142 for change in ({'user': 'other'}, {'resources': ['other']}, {'scopes': ['shale:read']}, {'resource': 'https://other.test/mcp/shale'}):
143 self.grant_as(**change)
144 with self.assertRaises(shale.Error):
145 self.create_label()
146 self.grant_as(user='other')
147 for column in ('access_issues', 'access_issues_submit', 'access_issues_comment'):
148 self.db.execute('UPDATE repositories SET ' + column + "='private' WHERE id=1"); self.db.commit()
149 with self.assertRaises(shale.Error):
150 self.call('create_issue' if column != 'access_issues_comment' else 'comment_issue', **({'title': 'Denied', 'description': '', 'status': 'todo'} if column != 'access_issues_comment' else {'id': 1, 'comment': 'Denied'}))
151 self.db.execute('UPDATE repositories SET ' + column + "='public' WHERE id=1"); self.db.commit()
152 self.grant_as()
153 with sqlite3.connect(self.root / 'connections.sqlite') as db:
154 db.execute("DELETE FROM records WHERE key='shale-session:owner'")
155 with self.assertRaises(shale.Error):
156 self.create_label()
157 self.assertEqual(self.db.execute('SELECT count(*) FROM issue_labels').fetchone(), (0,))
158
159 def test_missing_invalid_or_disabled_model_and_owner_fail_closed(self):
160 for model in (None, '', 'GPT 6', '../other', 'gpt-6\n', '<model-id>'):
161 with self.assertRaises(shale.Error):
162 self.call('create_label', name='bug', description='', color='#123456', model=model)
163 self.create_label()
164 with sqlite3.connect(self.root / 'accounts.sqlite') as db:
165 db.execute("UPDATE users SET profile=json_set(profile,'$.enabled',json('false')) WHERE username='ai-owner-gpt-6.1-sol'")
166 with self.assertRaises(shale.Error):
167 self.call('set_issue_status', id=1, status='done')
168 with sqlite3.connect(self.root / 'accounts.sqlite') as db:
169 db.execute("UPDATE users SET profile=json_set(profile,'$.enabled',json('false')) WHERE id='owner'")
170 with self.assertRaises(shale.Error):
171 self.call('create_issue', title='No fallback', description='', status='todo', model='claude-opus-5.5')
172 self.assertEqual(self.db.execute('SELECT count(*) FROM issue_actions').fetchone(), (0,))
173
174 def test_bad_ids_status_labels_and_rollback(self):
175 label = self.create_label()
176 for ids in ([label, label], [label, 999], [True], [-1]):
177 with self.assertRaises(shale.Error):
178 self.call('set_issue_labels', id=1, labels=ids)
179 for status in ('done', 'triage', None):
180 with self.assertRaises(shale.Error):
181 self.call('create_issue', title='Bad', description='', status=status)
182 with self.assertRaises(shale.Error):
183 self.call('set_issue_status', id=1, status='invented')
184 self.assertEqual(self.db.execute('SELECT count(*) FROM issue_actions').fetchone(), (0,))
185 self.assertEqual(self.db.execute('SELECT count(*) FROM issues').fetchone(), (3,))
186
187
188if __name__ == '__main__':
189 unittest.main()