File size: 3,207 Bytes
b296ad4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 | """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()
|