| """The MCP serializer must not emit a call to a different project.""" |
| import unittest |
| from tinyquery.evaluate import mcp_request,parse_action,score |
|
|
|
|
| class AdapterChecks(unittest.TestCase): |
| def test_execution_match_cannot_use_a_column_missing_from_context(self): |
| import json |
| from tinyquery.data import validate_sql |
| slots={'table':'items','parent':'items_groups','column':'name','numeric':'amount','category':'city', |
| 'date_column':'created_at','foreign':'group_id','label':'label','number':500,'upper':1000, |
| 'value':'Delhi','other':'Mumbai','date':'2025-01-15','year':2025,'month':1} |
| gold='SELECT name FROM items;';bad='SELECT name FROM items WHERE amount IS NULL OR amount IS NOT NULL;' |
| self.assertEqual(validate_sql(gold,'mysql',slots),validate_sql(bad,'mysql',slots)) |
| context={'schema':['CREATE TABLE items (id INTEGER, name TEXT);'],'tools':[{'name':'query','inputSchema':{ |
| 'type':'object','properties':{'query':{'type':'string'}},'required':['query']}}]} |
| row={'backend':'mysql','operation':'project','slots':slots,'context':context, |
| 'target':{'action':'call','name':'query','arguments':{'query':gold}}} |
| text=json.dumps({'action':'call','name':'query','arguments':{'query':bad}}) |
| self.assertFalse(score(row,text)['success']) |
|
|
| def test_unknown_table_or_write_is_not_serialized(self): |
| context={'backend':'mysql','schema':['CREATE TABLE customers (id INTEGER);'], |
| 'tools':[{'name':'query','description':'Execute a read-only MySQL SELECT query.', |
| 'inputSchema':{'type':'object','properties':{'sql':{'type':'string'}},'required':['sql']}}]} |
| action={'action':'call','name':'query','arguments':{'sql':'SELECT * FROM imaginary;'}} |
| with self.assertRaisesRegex(ValueError,'absent'):mcp_request(action,context) |
| action['arguments']['sql']='DELETE FROM customers;' |
| with self.assertRaisesRegex(ValueError,'read-only'):mcp_request(action,context) |
| action['arguments']['sql']='SELECT * FROM customers;' |
| self.assertEqual(mcp_request(action,context)['method'],'tools/call') |
|
|
| def test_duplicate_argument_keys_are_rejected(self): |
| with self.assertRaisesRegex(ValueError,'Duplicate JSON key'): |
| parse_action('{"action":"call","arguments":{"project_id":"a","project_id":"b"}}') |
|
|
| def test_wrong_project_is_rejected_without_repair(self): |
| context={'project_id':'demo_a','tools':[{'name':'execute_sql','inputSchema':{ |
| 'type':'object','properties':{'project_id':{'type':'string'},'query':{'type':'string'}}, |
| 'required':['project_id','query'],'additionalProperties':False}}]} |
| action={'action':'call','name':'execute_sql','arguments':{'project_id':'demo_b','query':'SELECT 1;'}} |
| with self.assertRaisesRegex(ValueError,'project scope'):mcp_request(action,context) |
| self.assertEqual(action['arguments']['project_id'],'demo_b') |
| action['arguments']['project_id']='demo_a' |
| self.assertEqual(mcp_request(action,context)['params']['arguments']['project_id'],'demo_a') |
|
|
|
|
| if __name__=='__main__':unittest.main() |
|
|
