import asyncio import copy import hashlib import json from importlib.resources import files import pytest from jsonschema import Draft202012Validator from viano_bug_analysis.analysis import CaseInterpreter, HypothesisBuilder from viano_bug_analysis.case import CaseBuilder from viano_bug_analysis.lifecycle import State from viano_bug_analysis.package import PackageReader, PackageWriter from viano_bug_analysis.profiles.vcore.db_tools import normalize_schema_descriptor from viano_bug_analysis.tools import InvestigationToolRouter, ToolError, ToolLimits from viano_bug_analysis.tools.database_query import DatabaseQueryTool from test_case_builder import case_setup from test_case_interpreter import gateway, FixtureAdapter from test_hypothesis_builder import ProposalAdapter @pytest.fixture def tool_case(case_setup, tmp_path): store, _, _, _, _, old_index = case_setup index = copy.deepcopy(old_index) index["context"]["source"].append("tool/code.cpp") index["context"]["database"] += ["tool/db.json", "tool/schema.json", "tool/raw.json"] snapshot = {"format":"viano-database-snapshot","format_version":1,"table":"switches", "columns":[{"name":"id","type":"integer"},{"name":"enabled","type":"boolean"},{"name":"ref","type":"text"}], "rows":[{"id":1,"enabled":False,"ref":"A"},{"id":2,"enabled":True,"ref":"A"},{"id":3,"enabled":False,"ref":"B"}],"coverage":"partial"} schema = {"table":"switches","columns":[{"column_name":"id","column_type":"int(11)"}]} incoming = tmp_path / "tools-input.zip" PackageWriter().write(store.latest("case-A","stable-package"), incoming, additions={"tool/code.cpp":b"void activate() {\n activate();\n activate_more();\n}\n", "tool/db.json":json.dumps(snapshot).encode(),"tool/schema.json":json.dumps(schema).encode(), "tool/raw.json":b'[{"id":1},{"id":"1"},{"id":true}]',"tool/index.json":json.dumps(index).encode()}, manifest_updates={"artifacts":{"case_inputs":"tool/index.json"}}) store.intake(incoming) CaseBuilder(store).build("case-A","stable-package") asyncio.run(CaseInterpreter(store,gateway(FixtureAdapter())).interpret("case-A","stable-package")) result = asyncio.run(HypothesisBuilder(store,gateway(ProposalAdapter())).build("case-A","stable-package")) return store,result.package def batch(operation="source.find_text", arguments=None): args = arguments if arguments is not None else {"source_refs":["tool/code.cpp"],"text":"activate","case_sensitive":True,"offset":0,"limit":10} return {"format":"viano-investigation-request","format_version":1,"case_id":"case-A","package_id":"stable-package","analysis_revision":1, "requests":[{"request_id":"Q1","hypothesis_id":"H1","action_id":"A1","operation":operation,"arguments":args}]} def query(tool_case, operation="source.find_text", args=None, router=None): return (router or InvestigationToolRouter()).query(tool_case[1],batch(operation,args))["results"][0]["data"] def test_literal_search_provenance_pagination_and_no_writes(tool_case): original = hashlib.sha256(tool_case[1].read_bytes()).hexdigest() args = batch()["requests"][0]["arguments"] args["limit"] = 1 first = query(tool_case,args=args) assert first["total_matches"] == 3 and first["truncated"] and first["next_offset"] == 1 assert first["matches"][0]["line"] == 1 with PackageReader(tool_case[1]) as reader: assert first["matches"][0]["sha256"] == hashlib.sha256(reader.read_bytes("tool/code.cpp")).hexdigest() args["offset"] = 2 last = query(tool_case,args=args) assert last["matches"][0]["line"] == 3 and not last["truncated"] assert hashlib.sha256(tool_case[1].read_bytes()).hexdigest() == original assert tool_case[0].latest("case-A","stable-package") == tool_case[1] def test_symbol_occurrences_not_substrings_or_definitions(tool_case): result = query(tool_case,"source.find_symbol",{"source_refs":["tool/code.cpp"],"symbol":"activate","offset":0,"limit":10}) assert [m["line"] for m in result["matches"]] == [1,2] assert result["interpretation"] == "text_occurrences_not_verified_definitions" def test_read_context_and_long_line_marking(tool_case): result = query(tool_case,"source.read_context",{"source_ref":"tool/code.cpp","start_line":2,"end_line":3}, InvestigationToolRouter(ToolLimits(max_line_characters=4))) assert [m["line"] for m in result["matches"]] == [2,3] assert all(m["text_truncated"] for m in result["matches"]) def test_database_structured_lookup_and_row_pointer(tool_case): result = query(tool_case,"database.lookup",{"source_ref":"tool/db.json","filters":{"enabled":False},"columns":["id"],"offset":0,"limit":1}) assert result["matches"][0]["row"] == {"id":1} assert result["matches"][0]["json_pointer"] == "/rows/0" assert result["coverage"] == "partial" and result["next_offset"] == 1 assert result["total_matches"] == 2 def test_database_references_are_explicit_values_not_relationship_claims(tool_case): result = query(tool_case,"database.find_references",{"source_ref":"tool/db.json","search_columns":["ref"],"value":"A","columns":[],"offset":0,"limit":10}) assert [m["row"]["id"] for m in result["matches"]] == [1,2] assert result["interpretation"] == "explicit_column_matches_not_inferred_relationships" @pytest.mark.parametrize("value,index", [(1,0),("1",1),(True,2)]) def test_database_equality_does_not_coerce_boolean_string_or_number(tool_case,value,index): result = query(tool_case,"database.lookup",{"source_ref":"tool/raw.json","filters":{"id":value},"columns":[],"offset":0,"limit":10}) assert result["total_matches"] == 1 and result["matches"][0]["json_pointer"] == f"/{index}" assert result["coverage"] == "unknown" def test_describe_generic_and_vcore_descriptor(tool_case): result = query(tool_case,"database.describe",{"source_ref":"tool/db.json"}) assert result["table"] == "switches" and len(result["columns"]) == 3 router = InvestigationToolRouter(database_tool=DatabaseQueryTool(normalize_schema_descriptor)) result = query(tool_case,"database.describe",{"source_ref":"tool/schema.json"},router) assert result["coverage"] == "schema_only" and result["columns"] == [{"name":"id","type":"int(11)"}] @pytest.mark.parametrize("change", [ lambda b: b["requests"][0].update(operation="shell.exec"), lambda b: b["requests"][0].update(operation="testing.request_human"), lambda b: b["requests"][0]["arguments"].update(sql="DROP TABLE switches"), lambda b: b["requests"][0]["arguments"].update(source_refs=["../escape"]), lambda b: b["requests"][0]["arguments"].update(source_refs=["C:/server/code.cpp"]), lambda b: b["requests"][0]["arguments"].update(source_refs=["tool/db.json"]), lambda b: b["requests"][0]["arguments"].update(source_refs=["unknown/vendor"]), lambda b: b["requests"][0]["arguments"].update(source_refs=["manifest.json"]), lambda b: b["requests"][0]["arguments"].update(source_refs=["tool/code.cpp","tool/code.cpp"]), lambda b: b.update(analysis_revision=2), lambda b: b["requests"][0].update(hypothesis_id="H99"), lambda b: b["requests"][0]["arguments"].update(limit=0), lambda b: b["requests"][0]["arguments"].update(limit=10.0), lambda b: b["requests"][0]["arguments"].update(text=""), ]) def test_unsafe_unsupported_and_wrongly_bound_requests_rejected(tool_case,change): request = batch() change(request) with pytest.raises(ToolError): InvestigationToolRouter().query(tool_case[1],request) def test_whole_batch_preflight_before_any_dispatch(tool_case,monkeypatch): router = InvestigationToolRouter() request = batch() invalid = batch("database.lookup",{"source_ref":"tool/db.json","filters":{"unknown":1},"columns":[],"offset":0,"limit":1})["requests"][0] invalid.update(request_id="Q2",action_id="A2") request["requests"].append(invalid) calls=[] monkeypatch.setattr(router.source,"execute",lambda *args: calls.append("dispatched")) with pytest.raises(ToolError,match="TOOL_DATABASE_COLUMN_UNKNOWN"): router.query(tool_case[1],request) assert not calls def test_search_empty_match_is_scoped_and_regex_is_literal(tool_case): args=batch()["requests"][0]["arguments"] args["text"]=".*" result=query(tool_case,args=args) assert result["matches"] == [] and result["scope"] == "packaged_source_snapshot" assert result["checked_sources"][0]["source_ref"] == "tool/code.cpp" def test_repeated_request_identity_rejected(tool_case): request=batch() request["requests"].append(copy.deepcopy(request["requests"][0])) with pytest.raises(ToolError,match="TOOL_DUPLICATE_REQUEST"): InvestigationToolRouter().query(tool_case[1],request) def test_context_ranges_and_operational_limits(tool_case): for start,end in [(0,1),(1,99),(3,2)]: with pytest.raises(ToolError): query(tool_case,"source.read_context",{"source_ref":"tool/code.cpp","start_line":start,"end_line":end}) for limits in [ToolLimits(max_file_bytes=2),ToolLimits(max_total_bytes=2),ToolLimits(max_matches=1),ToolLimits(max_request_bytes=2),ToolLimits(max_result_bytes=2)]: with pytest.raises(ToolError): query(tool_case,router=InvestigationToolRouter(limits)) def test_blocked_case_cannot_issue_queries(tool_case): store,_=tool_case path=store.advance("case-A","stable-package",State.BLOCKED,step_id="wait",reason="PROVIDER_UNAVAILABLE",retry_not_before="2099-01-01T00:00:00Z") with pytest.raises(ToolError,match="TOOL_PHASE"): InvestigationToolRouter().query(path,batch()) def test_new_undeclared_input_invalidates_query_context(tool_case,tmp_path): changed=tmp_path/"changed.zip" PackageWriter().write(tool_case[1],changed,additions={"new/raw":b"evidence"}) with pytest.raises(ToolError,match="TOOL_STALE_CASE"): InvestigationToolRouter().query(changed,batch()) @pytest.mark.parametrize("name", ["investigation-request","investigation-result","database-snapshot"]) def test_new_schemas_valid(name): Draft202012Validator.check_schema(json.loads(files("viano_bug_analysis.schemas").joinpath(name+'-v1.schema.json').read_text()))