190 lines
10 KiB
Python
190 lines
10 KiB
Python
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()))
|