519 lines
22 KiB
Python
519 lines
22 KiB
Python
#!/usr/bin/env python3
|
|
"""Run a document-ingestion-to-staged-proposal canary in disposable local Postgres."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
REPO_ROOT = HERE.parent
|
|
sys.path.insert(0, str(HERE))
|
|
|
|
import apply_proposal as ap # noqa: E402
|
|
import stage_normalized_proposal as normalized_stage # noqa: E402
|
|
|
|
SCHEMA = "livingip.workingLeoLocalIngestionProposalCanary.v1"
|
|
FIXTURE_SCHEMA = "livingip.workingLeoDocumentIngestionFixture.v1"
|
|
DEFAULT_FIXTURE = REPO_ROOT / "fixtures" / "working-leo" / "document-ingestion-v1.json"
|
|
CONTAINER_LABEL = "livingip.canary=working-leo-local-ingestion"
|
|
|
|
|
|
class CanaryError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def sha256_bytes(value: bytes) -> str:
|
|
return hashlib.sha256(value).hexdigest()
|
|
|
|
|
|
def canonical_sha256(value: Any) -> str:
|
|
return sha256_bytes(json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode())
|
|
|
|
|
|
def stable_uuid(artifact_sha256: str, label: str) -> str:
|
|
return str(uuid.uuid5(uuid.NAMESPACE_URL, f"livingip:working-leo-ingestion:{artifact_sha256}:{label}"))
|
|
|
|
|
|
def load_fixture(path: Path) -> tuple[dict[str, Any], dict[str, Any]]:
|
|
raw = path.read_bytes()
|
|
try:
|
|
fixture = json.loads(raw)
|
|
except json.JSONDecodeError as exc:
|
|
raise CanaryError(f"fixture is not valid JSON: {exc}") from exc
|
|
if not isinstance(fixture, dict) or fixture.get("schema") != FIXTURE_SCHEMA:
|
|
raise CanaryError(f"fixture schema must be {FIXTURE_SCHEMA}")
|
|
source = fixture.get("source")
|
|
extraction = fixture.get("extraction")
|
|
claim = extraction.get("claim") if isinstance(extraction, dict) else None
|
|
evidence = extraction.get("evidence") if isinstance(extraction, dict) else None
|
|
if not all(isinstance(item, dict) for item in (source, claim, evidence)):
|
|
raise CanaryError("fixture requires source, extraction.claim, and extraction.evidence objects")
|
|
required_source = ("identity", "source_key", "source_type", "title", "locator", "body", "metadata")
|
|
required_claim = ("claim_key", "type", "confidence", "text", "body", "metadata")
|
|
required_evidence = ("role", "body", "metadata")
|
|
for label, value, required in (
|
|
("source", source, required_source),
|
|
("claim", claim, required_claim),
|
|
("evidence", evidence, required_evidence),
|
|
):
|
|
missing = [key for key in required if value.get(key) in (None, "")]
|
|
if missing:
|
|
raise CanaryError(f"fixture {label} is missing: {', '.join(missing)}")
|
|
if not isinstance(value.get("metadata"), dict):
|
|
raise CanaryError(f"fixture {label}.metadata must be an object")
|
|
if claim["body"] not in source["body"]:
|
|
raise CanaryError("extracted claim body must be an exact substring of source.body")
|
|
if evidence["body"] not in source["body"]:
|
|
raise CanaryError("extracted evidence body must be an exact substring of source.body")
|
|
artifact_sha256 = sha256_bytes(raw)
|
|
source_content_sha256 = sha256_bytes(source["body"].encode())
|
|
return fixture, {
|
|
"path": str(path.resolve()),
|
|
"artifact_sha256": artifact_sha256,
|
|
"source_content_sha256": source_content_sha256,
|
|
"source_identity": source["identity"],
|
|
}
|
|
|
|
|
|
def build_row_ids(artifact_sha256: str) -> dict[str, str]:
|
|
return {
|
|
"source_capture_id": stable_uuid(artifact_sha256, "source-capture"),
|
|
"claim_extraction_id": stable_uuid(artifact_sha256, "claim-extraction"),
|
|
"evidence_extraction_id": stable_uuid(artifact_sha256, "evidence-extraction"),
|
|
"parent_proposal_id": stable_uuid(artifact_sha256, "rich-parent-proposal"),
|
|
}
|
|
|
|
|
|
def build_parent_packet(
|
|
fixture: dict[str, Any], input_receipt: dict[str, Any], row_ids: dict[str, str]
|
|
) -> dict[str, Any]:
|
|
source = fixture["source"]
|
|
claim = fixture["extraction"]["claim"]
|
|
evidence = fixture["extraction"]["evidence"]
|
|
parent_source_ref = (
|
|
f"local-ingestion:{input_receipt['artifact_sha256']}:"
|
|
f"source={row_ids['source_capture_id']}:claim={row_ids['claim_extraction_id']}:"
|
|
f"evidence={row_ids['evidence_extraction_id']}"
|
|
)
|
|
return {
|
|
"id": row_ids["parent_proposal_id"],
|
|
"proposal_type": "attach_evidence",
|
|
"status": "pending_review",
|
|
"proposed_by_handle": "leo",
|
|
"proposed_by_agent_id": None,
|
|
"channel": "local_ingestion_canary",
|
|
"source_ref": parent_source_ref,
|
|
"rationale": "Stage one hash-bound claim/evidence extraction from a captured local document fixture.",
|
|
"payload": {
|
|
"claim_candidates": [
|
|
{
|
|
"claim_key": claim["claim_key"],
|
|
"type": claim["type"],
|
|
"confidence": claim["confidence"],
|
|
"text": claim["text"],
|
|
"body": claim["body"],
|
|
"evidence_tier": "isolated_fixture",
|
|
"needs_research": claim["metadata"].get("needs_research", False),
|
|
"evidence": [{"source_key": source["source_key"], "role": evidence["role"]}],
|
|
"edges": [],
|
|
}
|
|
],
|
|
"source_candidates": [
|
|
{
|
|
"source_key": source["source_key"],
|
|
"source_type": source["source_type"],
|
|
"title": source["title"],
|
|
"storage_path": input_receipt["path"],
|
|
"url": None,
|
|
"source_quality": "local realistic ingestion fixture",
|
|
"key_excerpt": evidence["body"],
|
|
"content_sha256": input_receipt["source_content_sha256"],
|
|
}
|
|
],
|
|
"dedupe_conflict_assessment": {
|
|
"database_search_query": claim["claim_key"].replace("_", " "),
|
|
"potential_existing_claim_ids": [],
|
|
"conflicts": [],
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
def build_schema_sql() -> str:
|
|
return r"""\set ON_ERROR_STOP on
|
|
create schema kb_canary;
|
|
create schema kb_stage;
|
|
create table kb_canary.source_captures (
|
|
id uuid primary key,
|
|
source_identity text not null unique,
|
|
source_type text not null,
|
|
title text not null,
|
|
locator text not null,
|
|
artifact_path text not null,
|
|
artifact_sha256 text not null check (artifact_sha256 ~ '^[0-9a-f]{64}$'),
|
|
content_sha256 text not null check (content_sha256 ~ '^[0-9a-f]{64}$'),
|
|
body text not null,
|
|
metadata jsonb not null
|
|
);
|
|
create table kb_canary.claim_extractions (
|
|
id uuid primary key,
|
|
source_capture_id uuid not null references kb_canary.source_captures(id),
|
|
claim_key text not null,
|
|
body text not null,
|
|
metadata jsonb not null
|
|
);
|
|
create table kb_canary.evidence_extractions (
|
|
id uuid primary key,
|
|
claim_extraction_id uuid not null references kb_canary.claim_extractions(id),
|
|
source_capture_id uuid not null references kb_canary.source_captures(id),
|
|
role text not null,
|
|
body text not null,
|
|
metadata jsonb not null
|
|
);
|
|
create table kb_stage.kb_proposals (
|
|
id uuid primary key,
|
|
proposal_type text not null,
|
|
status text not null,
|
|
proposed_by_handle text,
|
|
proposed_by_agent_id uuid,
|
|
channel text not null,
|
|
source_ref text not null,
|
|
rationale text not null,
|
|
payload jsonb not null,
|
|
reviewed_by_handle text,
|
|
reviewed_by_agent_id uuid,
|
|
reviewed_at timestamptz,
|
|
review_note text,
|
|
applied_by_handle text,
|
|
applied_by_agent_id uuid,
|
|
applied_at timestamptz
|
|
);
|
|
"""
|
|
|
|
|
|
def build_capture_sql(fixture: dict[str, Any], input_receipt: dict[str, Any], row_ids: dict[str, str]) -> str:
|
|
source = fixture["source"]
|
|
claim = fixture["extraction"]["claim"]
|
|
evidence = fixture["extraction"]["evidence"]
|
|
q = ap.sql_literal
|
|
claim_metadata = {
|
|
**claim["metadata"],
|
|
"claim_type": claim["type"],
|
|
"confidence": claim["confidence"],
|
|
"text": claim["text"],
|
|
}
|
|
return rf"""\set ON_ERROR_STOP on
|
|
begin;
|
|
insert into kb_canary.source_captures
|
|
(id, source_identity, source_type, title, locator, artifact_path,
|
|
artifact_sha256, content_sha256, body, metadata)
|
|
values
|
|
({q(row_ids["source_capture_id"])}::uuid, {q(source["identity"])}, {q(source["source_type"])},
|
|
{q(source["title"])}, {q(source["locator"])}, {q(input_receipt["path"])},
|
|
{q(input_receipt["artifact_sha256"])}, {q(input_receipt["source_content_sha256"])},
|
|
{q(source["body"])}, {q(json.dumps(source["metadata"], sort_keys=True))}::jsonb);
|
|
insert into kb_canary.claim_extractions
|
|
(id, source_capture_id, claim_key, body, metadata)
|
|
values
|
|
({q(row_ids["claim_extraction_id"])}::uuid, {q(row_ids["source_capture_id"])}::uuid,
|
|
{q(claim["claim_key"])}, {q(claim["body"])}, {q(json.dumps(claim_metadata, sort_keys=True))}::jsonb);
|
|
insert into kb_canary.evidence_extractions
|
|
(id, claim_extraction_id, source_capture_id, role, body, metadata)
|
|
values
|
|
({q(row_ids["evidence_extraction_id"])}::uuid, {q(row_ids["claim_extraction_id"])}::uuid,
|
|
{q(row_ids["source_capture_id"])}::uuid, {q(evidence["role"])}, {q(evidence["body"])},
|
|
{q(json.dumps(evidence["metadata"], sort_keys=True))}::jsonb);
|
|
commit;
|
|
"""
|
|
|
|
|
|
def build_readback_sql(row_ids: dict[str, str], proposal_id: str) -> str:
|
|
q = ap.sql_literal
|
|
return rf"""\set ON_ERROR_STOP on
|
|
begin transaction read only;
|
|
select jsonb_build_object(
|
|
'source_capture', (select to_jsonb(s) from kb_canary.source_captures s
|
|
where id = {q(row_ids["source_capture_id"])}::uuid),
|
|
'claim_extraction', (select to_jsonb(c) from kb_canary.claim_extractions c
|
|
where id = {q(row_ids["claim_extraction_id"])}::uuid),
|
|
'evidence_extraction', (select to_jsonb(e) from kb_canary.evidence_extractions e
|
|
where id = {q(row_ids["evidence_extraction_id"])}::uuid),
|
|
'staged_proposal', (select to_jsonb(p) from kb_stage.kb_proposals p
|
|
where id = {q(proposal_id)}::uuid),
|
|
'counts', jsonb_build_object(
|
|
'source_captures', (select count(*) from kb_canary.source_captures),
|
|
'claim_extractions', (select count(*) from kb_canary.claim_extractions),
|
|
'evidence_extractions', (select count(*) from kb_canary.evidence_extractions),
|
|
'staged_proposals', (select count(*) from kb_stage.kb_proposals)
|
|
)
|
|
)::text;
|
|
rollback;
|
|
"""
|
|
|
|
|
|
class DockerPostgres:
|
|
def __init__(self, name: str) -> None:
|
|
self.name = name
|
|
self.docker = shutil.which("docker")
|
|
if not self.docker:
|
|
raise CanaryError("docker executable is unavailable")
|
|
self.started = False
|
|
self.volume_mounts: list[dict[str, Any]] = []
|
|
|
|
def _run(
|
|
self, args: list[str], *, input_text: str | None = None, check: bool = True
|
|
) -> subprocess.CompletedProcess[str]:
|
|
result = subprocess.run([self.docker, *args], input=input_text, text=True, capture_output=True, check=False)
|
|
if check and result.returncode != 0:
|
|
detail = (result.stderr or result.stdout).strip()
|
|
raise CanaryError(f"docker {' '.join(args[:3])} failed ({result.returncode}): {detail}")
|
|
return result
|
|
|
|
def start(self) -> dict[str, Any]:
|
|
result = self._run(
|
|
[
|
|
"run",
|
|
"--rm",
|
|
"-d",
|
|
"--name",
|
|
self.name,
|
|
"--network",
|
|
"none",
|
|
"--label",
|
|
CONTAINER_LABEL,
|
|
"--tmpfs",
|
|
"/var/lib/postgresql/data:rw,nosuid,size=256m",
|
|
"-e",
|
|
"POSTGRES_HOST_AUTH_METHOD=trust",
|
|
"-e",
|
|
"POSTGRES_DB=teleo",
|
|
"postgres:16-alpine",
|
|
]
|
|
)
|
|
self.started = True
|
|
container_id = result.stdout.strip()
|
|
deadline = time.monotonic() + 30
|
|
ready_streak = 0
|
|
while time.monotonic() < deadline:
|
|
ready = self._run(["exec", self.name, "pg_isready", "-U", "postgres", "-d", "teleo"], check=False)
|
|
if ready.returncode == 0:
|
|
ready_streak += 1
|
|
if ready_streak >= 2:
|
|
break
|
|
else:
|
|
ready_streak = 0
|
|
time.sleep(0.25)
|
|
else:
|
|
raise CanaryError("disposable Postgres did not become ready within 30 seconds")
|
|
inspect = json.loads(self._run(["inspect", self.name]).stdout)[0]
|
|
if inspect["HostConfig"]["NetworkMode"] != "none":
|
|
raise CanaryError("disposable Postgres must use network mode none")
|
|
if inspect["Config"]["Labels"].get("livingip.canary") != "working-leo-local-ingestion":
|
|
raise CanaryError("disposable Postgres is missing its canary label")
|
|
self.volume_mounts = [
|
|
{
|
|
"type": mount.get("Type"),
|
|
"name": mount.get("Name"),
|
|
"destination": mount.get("Destination"),
|
|
}
|
|
for mount in inspect.get("Mounts", [])
|
|
if mount.get("Type") == "volume"
|
|
]
|
|
if self.volume_mounts:
|
|
raise CanaryError("disposable Postgres unexpectedly created a Docker volume")
|
|
return {
|
|
"container_id": container_id,
|
|
"name": self.name,
|
|
"image": inspect["Config"]["Image"],
|
|
"network_mode": inspect["HostConfig"]["NetworkMode"],
|
|
"tmpfs_data_dir": "/var/lib/postgresql/data" in inspect["HostConfig"].get("Tmpfs", {}),
|
|
"docker_volume_mounts": self.volume_mounts,
|
|
"canary_label": inspect["Config"]["Labels"].get("livingip.canary"),
|
|
}
|
|
|
|
def psql(self, sql: str) -> str:
|
|
result = self._run(
|
|
["exec", "-i", self.name, "psql", "-U", "postgres", "-d", "teleo", "-Atq"],
|
|
input_text=sql,
|
|
)
|
|
return result.stdout.strip()
|
|
|
|
def psql_json(self, sql: str) -> dict[str, Any]:
|
|
output = self.psql(sql)
|
|
lines = [line for line in output.splitlines() if line.strip()]
|
|
if not lines:
|
|
raise CanaryError("Postgres readback returned no JSON")
|
|
try:
|
|
value = json.loads(lines[-1])
|
|
except json.JSONDecodeError as exc:
|
|
raise CanaryError(f"Postgres readback was not JSON: {lines[-1]}") from exc
|
|
if not isinstance(value, dict):
|
|
raise CanaryError("Postgres readback must be a JSON object")
|
|
return value
|
|
|
|
def cleanup(self) -> dict[str, Any]:
|
|
remove = self._run(["rm", "-f", self.name], check=False) if self.started else None
|
|
inspect = self._run(["inspect", self.name], check=False)
|
|
return {
|
|
"remove_attempted": self.started,
|
|
"remove_returncode": remove.returncode if remove else None,
|
|
"container_absent": inspect.returncode != 0,
|
|
"docker_volume_mounts_observed": self.volume_mounts,
|
|
"anonymous_or_named_volume_created": bool(self.volume_mounts),
|
|
}
|
|
|
|
|
|
def proposal_links(readback: dict[str, Any], input_receipt: dict[str, Any]) -> dict[str, Any]:
|
|
proposal = readback["staged_proposal"]
|
|
apply_payload = proposal["payload"]["apply_payload"]
|
|
source_rows = apply_payload["sources"]
|
|
evidence_rows = apply_payload["evidence"]
|
|
return {
|
|
"proposal_id": proposal["id"],
|
|
"proposal_status": proposal["status"],
|
|
"proposal_source_ref": proposal["source_ref"],
|
|
"planned_claim_ids": [row["id"] for row in apply_payload["claims"]],
|
|
"planned_source_ids": [row["id"] for row in source_rows],
|
|
"planned_evidence_keys": [
|
|
{"claim_id": row["claim_id"], "source_id": row["source_id"], "role": row["role"]} for row in evidence_rows
|
|
],
|
|
"source_content_hash_matches": any(
|
|
row.get("hash") == input_receipt["source_content_sha256"] for row in source_rows
|
|
),
|
|
"parent_source_ref_embedded": any(
|
|
readback["source_capture"]["artifact_sha256"] in str(row.get("excerpt"))
|
|
and readback["source_capture"]["id"] in str(row.get("excerpt"))
|
|
and readback["claim_extraction"]["id"] in str(row.get("excerpt"))
|
|
and readback["evidence_extraction"]["id"] in str(row.get("excerpt"))
|
|
for row in source_rows
|
|
),
|
|
}
|
|
|
|
|
|
def evaluate_checks(
|
|
fixture: dict[str, Any], input_receipt: dict[str, Any], row_ids: dict[str, str], readback: dict[str, Any]
|
|
) -> dict[str, bool]:
|
|
source = readback.get("source_capture") or {}
|
|
claim = readback.get("claim_extraction") or {}
|
|
evidence = readback.get("evidence_extraction") or {}
|
|
proposal = readback.get("staged_proposal") or {}
|
|
links = proposal_links(readback, input_receipt)
|
|
return {
|
|
"input_artifact_hash_persisted": source.get("artifact_sha256") == input_receipt["artifact_sha256"],
|
|
"source_content_hash_persisted": source.get("content_sha256") == input_receipt["source_content_sha256"],
|
|
"source_identity_persisted": source.get("source_identity") == input_receipt["source_identity"],
|
|
"claim_links_to_source_capture": claim.get("source_capture_id") == row_ids["source_capture_id"],
|
|
"claim_body_and_metadata_persisted": (
|
|
claim.get("body") == fixture["extraction"]["claim"]["body"]
|
|
and claim.get("metadata", {}).get("text") == fixture["extraction"]["claim"]["text"]
|
|
),
|
|
"evidence_links_to_claim_and_source": (
|
|
evidence.get("claim_extraction_id") == row_ids["claim_extraction_id"]
|
|
and evidence.get("source_capture_id") == row_ids["source_capture_id"]
|
|
),
|
|
"evidence_body_and_metadata_persisted": (
|
|
evidence.get("body") == fixture["extraction"]["evidence"]["body"]
|
|
and evidence.get("metadata") == fixture["extraction"]["evidence"]["metadata"]
|
|
),
|
|
"one_row_per_ingestion_stage": readback.get("counts")
|
|
== {"source_captures": 1, "claim_extractions": 1, "evidence_extractions": 1, "staged_proposals": 1},
|
|
"proposal_is_pending_review": proposal.get("status") == "pending_review",
|
|
"proposal_has_source_ref": bool(proposal.get("source_ref")),
|
|
"proposal_payload_contains_input_hash": links["source_content_hash_matches"],
|
|
"proposal_payload_contains_capture_row_linkage": links["parent_source_ref_embedded"],
|
|
"no_review_or_apply_metadata": all(
|
|
proposal.get(key) is None
|
|
for key in ("reviewed_by_handle", "reviewed_at", "applied_by_handle", "applied_at")
|
|
),
|
|
}
|
|
|
|
|
|
def run_canary(fixture_path: Path, output: Path | None = None) -> dict[str, Any]:
|
|
fixture, input_receipt = load_fixture(fixture_path)
|
|
row_ids = build_row_ids(input_receipt["artifact_sha256"])
|
|
parent = build_parent_packet(fixture, input_receipt, row_ids)
|
|
child = normalized_stage.prepare_normalized_child(parent)
|
|
container_name = f"working-leo-ingest-{uuid.uuid4().hex[:12]}"
|
|
postgres = DockerPostgres(container_name)
|
|
receipt: dict[str, Any] = {
|
|
"schema": SCHEMA,
|
|
"status": "fail",
|
|
"required_tier": "isolated_realistic_ingestion_to_proposal",
|
|
"input": input_receipt,
|
|
"row_ids": row_ids,
|
|
"production_mutation_attempted": False,
|
|
"canonical_apply_attempted": False,
|
|
}
|
|
try:
|
|
receipt["environment"] = postgres.start()
|
|
postgres.psql(build_schema_sql())
|
|
postgres.psql(build_capture_sql(fixture, input_receipt, row_ids))
|
|
stage_result = postgres.psql_json(normalized_stage.build_stage_sql(child))
|
|
readback = postgres.psql_json(build_readback_sql(row_ids, child["id"]))
|
|
checks = evaluate_checks(fixture, input_receipt, row_ids, readback)
|
|
receipt.update(
|
|
{
|
|
"stage_command_result": stage_result,
|
|
"rows": readback,
|
|
"row_link_fields_proven": proposal_links(readback, input_receipt),
|
|
"checks": checks,
|
|
"status": "pass" if all(checks.values()) else "fail",
|
|
}
|
|
)
|
|
except (CanaryError, OSError, ValueError, KeyError) as exc:
|
|
receipt["error"] = {"type": type(exc).__name__, "message": str(exc)}
|
|
finally:
|
|
receipt["cleanup"] = postgres.cleanup()
|
|
receipt["cleanup"]["network_mode_none"] = (
|
|
receipt.get("environment", {}).get("network_mode") == "none"
|
|
)
|
|
receipt["cleanup"]["no_production_target_referenced"] = receipt[
|
|
"cleanup"
|
|
]["network_mode_none"]
|
|
if not all(
|
|
(
|
|
receipt["cleanup"]["container_absent"],
|
|
receipt["cleanup"]["network_mode_none"],
|
|
)
|
|
):
|
|
receipt["status"] = "fail"
|
|
receipt["strongest_claim"] = (
|
|
"A realistic local document fixture was hash-captured, replayed through deterministic claim/evidence "
|
|
"extraction, normalized by the existing rich-proposal normalizer, and staged/read back as one "
|
|
"pending_review proposal in disposable networkless Postgres."
|
|
)
|
|
receipt["residual_gap"] = (
|
|
"This does not prove live generative extraction, arbitrary document coverage, canonical apply, or any "
|
|
"hosted/production runtime."
|
|
)
|
|
if output:
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
output.write_text(json.dumps(receipt, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
|
return receipt
|
|
|
|
|
|
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--fixture", type=Path, default=DEFAULT_FIXTURE)
|
|
parser.add_argument("--output", type=Path)
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = parse_args(argv)
|
|
receipt = run_canary(args.fixture.resolve(), args.output.resolve() if args.output else None)
|
|
print(json.dumps(receipt, sort_keys=True, separators=(",", ":")))
|
|
return 0 if receipt["status"] == "pass" else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|