Task 2. Xây dựng dịch vụ parser CPG tăng dần

import os
import json
import shutil
import subprocess
import hashlib
import sqlite3
from pathlib import Path
from collections import Counter
import jsonschema

# Resolve PROJECT_ROOT
PROJECT_ROOT = Path(
    subprocess.run(
        ["git", "rev-parse", "--show-toplevel"],
        check=True,
        capture_output=True,
        text=True,
    ).stdout.strip()
)

# Helper to run CLI commands and check return codes
def run_command(
    args: list[str],
    *,
    expected_return_codes: set[int] | None = None,
    env: dict[str, str] | None = None,
) -> subprocess.CompletedProcess[str]:
    if expected_return_codes is None:
        expected_return_codes = {0}
    
    result = subprocess.run(
        args,
        capture_output=True,
        text=True,
        cwd=str(PROJECT_ROOT),
        env=env
    )
    if result.returncode not in expected_return_codes:
        print("STDOUT:", result.stdout)
        print("STDERR:", result.stderr)
        raise RuntimeError(
            f"Command {args} failed with exit code {result.returncode}. Expected one of {expected_return_codes}"
        )
    return result

# Isolated paths for demonstration - shifted to runtime temporary workspace (workspace/tmp/notebook/)
PARSER_SMOKE_STATE = PROJECT_ROOT / "workspace/tmp/notebook/parser_smoke.sqlite3"
NOTEBOOK_OUTPUT_DIRECTORY = PROJECT_ROOT / "workspace/tmp/notebook-parser"
SCHEMA_DIR = PROJECT_ROOT / "schemas"
SOURCE_REPOSITORY = PROJECT_ROOT / "workspace/source/transformers-pr-agent"

# Reset isolated demonstration state and dirs in temporary workspace
NOTEBOOK_STATE_DIRECTORY = PROJECT_ROOT / "workspace/tmp/notebook"
if NOTEBOOK_STATE_DIRECTORY.exists():
    shutil.rmtree(NOTEBOOK_STATE_DIRECTORY, ignore_errors=True)
NOTEBOOK_STATE_DIRECTORY.mkdir(parents=True, exist_ok=True)

if NOTEBOOK_OUTPUT_DIRECTORY.exists():
    shutil.rmtree(NOTEBOOK_OUTPUT_DIRECTORY, ignore_errors=True)
NOTEBOOK_OUTPUT_DIRECTORY.mkdir(parents=True, exist_ok=True)

print("PROJECT_ROOT:", PROJECT_ROOT)
print("Isolated state DB:", PARSER_SMOKE_STATE)
print("Isolated output dir:", NOTEBOOK_OUTPUT_DIRECTORY / "smoke")
print("Schemas directory:", SCHEMA_DIR)
print("SOURCE_REPOSITORY:", SOURCE_REPOSITORY)

# Initial parse run on 3 files using the official CLI
env = {**os.environ, "PARSER_STATE_DB": str(PARSER_SMOKE_STATE)}

result = run_command(
    [
        "uv", "run", "lab04", "parse-repository",
        "--scope", "smoke",
        "--limit", "3",
        "--dry-run",
        "--clean-output",
        "--out-dir", str(NOTEBOOK_OUTPUT_DIRECTORY / "smoke"),
    ],
    env=env
)
print("CLI command executed successfully!")
print(result.stdout)

def count_jsonl(path: Path) -> int:
    if not path.exists():
        return 0
    with path.open(encoding="utf-8") as handle:
        return sum(1 for _ in handle)

smoke_out = NOTEBOOK_OUTPUT_DIRECTORY / "smoke"
counts = {
    name: count_jsonl(smoke_out / f"{name}.jsonl")
    for name in ["nodes", "edges", "metadata", "errors"]
}

for name, count in counts.items():
    print(f"{name}: {count}")

with (smoke_out / "nodes.jsonl").open(encoding="utf-8") as handle:
    sample_node = json.loads(next(handle))

print(json.dumps(sample_node, indent=2, ensure_ascii=False))

edge_types = Counter()
first_edge = None
with (smoke_out / "edges.jsonl").open(encoding="utf-8") as handle:
    for line in handle:
        event = json.loads(line)
        first_edge = first_edge or event
        edge_types[event["edge"]["edge_type"]] += 1

print("Edge types:")
for edge_type, count in edge_types.most_common():
    print(f"{edge_type}: {count}")

print("\nSample edge:")
print(json.dumps(first_edge, indent=2, ensure_ascii=False))

with (smoke_out / "metadata.jsonl").open(encoding="utf-8") as handle:
    sample_metadata = json.loads(next(handle))

print(json.dumps(sample_metadata, indent=2, ensure_ascii=False))

# Schema validation helper using root schemas
def validate_event_with_schema(event: dict, schema_name: str):
    schema_path = SCHEMA_DIR / schema_name
    with open(schema_path, "r", encoding="utf-8") as f:
        schema = json.load(f)
    jsonschema.Draft202012Validator(schema).validate(event)

# Validate all nodes, edges, and metadata
event_types_checked = set()

# Check Node schemas
with (smoke_out / "nodes.jsonl").open(encoding="utf-8") as f:
    for line in f:
        event = json.loads(line)
        validate_event_with_schema(event, "node-event.schema.json")
        event_types_checked.add(event["event_type"])
        
# Check Edge schemas
with (smoke_out / "edges.jsonl").open(encoding="utf-8") as f:
    for line in f:
        event = json.loads(line)
        validate_event_with_schema(event, "edge-event.schema.json")
        event_types_checked.add(event["event_type"])
        
# Check Metadata schemas
with (smoke_out / "metadata.jsonl").open(encoding="utf-8") as f:
    for line in f:
        event = json.loads(line)
        validate_event_with_schema(event, "metadata-event.schema.json")
        event_types_checked.add(event["event_type"])

print("JSON Schema validation using root schemas directory PASSED successfully!")
print("Validated event types:", list(event_types_checked))

# Stable IDs and content hashes determinism verification
determinism_primary_db = PROJECT_ROOT / "workspace/tmp/notebook/parser_determinism_primary.sqlite3"
determinism_reference_db = PROJECT_ROOT / "workspace/tmp/notebook/parser_determinism_reference.sqlite3"

if determinism_primary_db.exists():
    determinism_primary_db.unlink()
if determinism_reference_db.exists():
    determinism_reference_db.unlink()
    
# Copy the smoke DB to primary determinism DB
shutil.copy(PARSER_SMOKE_STATE, determinism_primary_db)

state_a = NOTEBOOK_OUTPUT_DIRECTORY / "determinism-primary"
state_b = NOTEBOOK_OUTPUT_DIRECTORY / "determinism-reference"

# Copy smoke outputs to determinism-primary
if state_a.exists():
    shutil.rmtree(state_a)
shutil.copytree(NOTEBOOK_OUTPUT_DIRECTORY / "smoke", state_a)

env_b = {**os.environ, "PARSER_STATE_DB": str(determinism_reference_db)}

run_command(
    [
        "uv", "run", "lab04", "parse-repository",
        "--scope", "smoke",
        "--limit", "3",
        "--dry-run",
        "--clean-output",
        "--out-dir", str(state_b),
    ],
    env=env_b
)

def get_ids_and_hashes(out_dir_path: Path):
    node_ids = set()
    edge_ids = set()
    file_ids = set()
    hashes = set()
    
    with (out_dir_path / "nodes.jsonl").open(encoding="utf-8") as f:
        for line in f:
            evt = json.loads(line)
            node_ids.add(evt["node"]["node_id"])
    with (out_dir_path / "edges.jsonl").open(encoding="utf-8") as f:
        for line in f:
            evt = json.loads(line)
            edge_ids.add(evt["edge"]["edge_id"])
    with (out_dir_path / "metadata.jsonl").open(encoding="utf-8") as f:
        for line in f:
            evt = json.loads(line)
            file_ids.add(evt["file_id"])
            hashes.add(evt["content_hash"])
    return node_ids, edge_ids, file_ids, hashes

node_ids_a, edge_ids_a, file_ids_a, hashes_a = get_ids_and_hashes(state_a)
node_ids_b, edge_ids_b, file_ids_b, hashes_b = get_ids_and_hashes(state_b)

assert node_ids_a == node_ids_b, "Node IDs are not deterministic"
assert edge_ids_a == edge_ids_b, "Edge IDs are not deterministic"
assert file_ids_a == file_ids_b, "File IDs are not deterministic"
assert hashes_a == hashes_b, "Content hashes are not deterministic"

print("Stable IDs and content hashes determinism verification A/B PASSED successfully!")

# Incremental skip verification (re-parsing same repo with same state DB)
env_skip = {**os.environ, "PARSER_STATE_DB": str(PARSER_SMOKE_STATE)}
skip_out = NOTEBOOK_OUTPUT_DIRECTORY / "unchanged"

result_skip = run_command(
    [
        "uv", "run", "lab04", "parse-repository",
        "--scope", "smoke",
        "--limit", "3",
        "--dry-run",
        "--clean-output",
        "--out-dir", str(skip_out),
    ],
    env=env_skip
)

print(result_skip.stdout)
assert count_jsonl(skip_out / "nodes.jsonl") == 0, "No new nodes should be generated"
assert count_jsonl(skip_out / "edges.jsonl") == 0, "No new edges should be generated"
assert count_jsonl(skip_out / "metadata.jsonl") == 0, "No new metadata should be generated"

print("Incremental unchanged-file skip verification PASSED successfully!")

# Parser error verification using broken syntax fixture
parser_error_db = PROJECT_ROOT / "workspace/tmp/notebook/parser_error.sqlite3"
if parser_error_db.exists():
    parser_error_db.unlink()
    
# Copy fixture to target repo to comply with CLI path resolution logic
target_fixture_dir = SOURCE_REPOSITORY / "tests/fixtures"
target_fixture_dir.mkdir(parents=True, exist_ok=True)
shutil.copy(PROJECT_ROOT / "tests/fixtures/broken_syntax.py", target_fixture_dir / "broken_syntax.py")

env_err = {**os.environ, "PARSER_STATE_DB": str(parser_error_db)}
syntax_error_out = NOTEBOOK_OUTPUT_DIRECTORY / "syntax-error"

# Run command expecting return code 1
result_err = run_command(
    [
        "uv", "run", "lab04", "parse-file",
        "--file", "tests/fixtures/broken_syntax.py",
        "--dry-run",
        "--clean-output",
        "--out-dir", str(syntax_error_out),
    ],
    expected_return_codes={1},
    env=env_err
)

# Cleanup copied fixture immediately
if (target_fixture_dir / "broken_syntax.py").exists():
    (target_fixture_dir / "broken_syntax.py").unlink()

# --- Verification & Presentation ---

# 1. Assert invalid syntax file was rejected (returncode == 1)
assert result_err.returncode == 1, "CLI must return exit code 1 on parse failure"

# 2. Verify errors.jsonl exists and conforms to schema
assert (syntax_error_out / "errors.jsonl").exists(), "errors.jsonl must be created"
with (syntax_error_out / "errors.jsonl").open(encoding="utf-8") as f:
    err_event = json.loads(next(f))

# Assert event type is PARSER_ERROR
assert err_event["event_type"] == "PARSER_ERROR", "Event type must be PARSER_ERROR"

# Validate schema
validate_event_with_schema(err_event, "error-event.schema.json")

# 3. Assert error event contains correct file path and content hash
with open(PROJECT_ROOT / "tests/fixtures/broken_syntax.py", "rb") as f:
    expected_content = f.read()
expected_hash = hashlib.sha256(expected_content).hexdigest()

assert err_event["file_path"] == "tests/fixtures/broken_syntax.py", f"Incorrect file path: {err_event['file_path']}"
assert err_event["content_hash"] == expected_hash, f"Incorrect content hash: {err_event['content_hash']}"

# 4. Assert no graph success state was committed to SQLite state store
conn = sqlite3.connect(parser_error_db)
cursor = conn.cursor()
cursor.execute("SELECT * FROM file_state WHERE file_path LIKE '%broken_syntax.py%'")
rows = cursor.fetchall()
conn.close()
assert len(rows) == 0, "Error: Graph state must not be committed to SQLite for files with syntax errors!"

# 5. Assert no node/edge events were generated
assert not (syntax_error_out / "nodes.jsonl").exists(), "nodes.jsonl must not be created for error file"
assert not (syntax_error_out / "edges.jsonl").exists(), "edges.jsonl must not be created for error file"

# Print outputs as specified
print("Parser execution result:")
print("FAILED AS EXPECTED")
print()
print("Verification result:")
print("PASSED")
print()
print("## Kết quả Parser")
print("- Input: `tests/fixtures/broken_syntax.py`")
print("- Parser status: `FAILED`")
print("- CLI exit code: `1`")
print("- Expected behavior: `Yes`")
print()
print("## Kết quả Verification")
print("[PASS] Parser rejected the invalid Python file as expected.")
print("[PASS] CLI returned a non-zero exit code as expected.")
print("[PASS] PARSER_ERROR event was generated.")
print("[PASS] Error event passed JSON Schema validation.")
print("[PASS] No successful graph state was committed.")