from __future__ import annotations

import ast
import importlib.util
import json
import pathlib
from dataclasses import dataclass
from typing import Any, Iterable


SENSITIVE_NAMES = {
    "subprocess", "multiprocessing", "threading", "signal", "socket",
    "ProcessPoolExecutor", "ThreadPoolExecutor", "Popen", "fork", "spawn",
    "chdir", "putenv", "unsetenv", "lru_cache", "cache",
}
SENSITIVE_ATTRIBUTES = {
    "os.environ", "os.chdir", "os.fork", "os.putenv", "os.unsetenv",
    "subprocess.Popen", "subprocess.run", "subprocess.call",
    "multiprocessing.Process", "threading.Thread", "signal.signal",
    "socket.socket", "concurrent.futures.ThreadPoolExecutor",
    "concurrent.futures.ProcessPoolExecutor", "functools.lru_cache", "functools.cache",
}


def _attribute_name(node: ast.AST) -> str:
    parts: list[str] = []
    while isinstance(node, ast.Attribute):
        parts.append(node.attr)
        node = node.value
    if isinstance(node, ast.Name):
        parts.append(node.id)
    return ".".join(reversed(parts))


def _node_hazards(node: ast.AST) -> set[str]:
    hazards: set[str] = set()
    for item in ast.walk(node):
        if isinstance(item, (ast.Import, ast.ImportFrom)):
            names = [alias.name for alias in item.names]
            for name in names:
                if name.split(".")[0] in SENSITIVE_NAMES:
                    hazards.add("import:" + name)
        if isinstance(item, ast.Call):
            name = _attribute_name(item.func)
            if name in SENSITIVE_ATTRIBUTES or name.rsplit(".", 1)[-1] in SENSITIVE_NAMES:
                hazards.add("call:" + name)
        if isinstance(item, (ast.Assign, ast.AugAssign, ast.AnnAssign)):
            targets = item.targets if isinstance(item, ast.Assign) else [item.target]
            for target in targets:
                name = _attribute_name(target)
                if name.startswith("os.environ"):
                    hazards.add("mutation:" + name)
        if isinstance(item, ast.Subscript) and _attribute_name(item.value) == "os.environ":
            hazards.add("environment_subscript")
    return hazards


def _module_path(root: pathlib.Path, test_id: str) -> pathlib.Path:
    module_name = ".".join(test_id.split(".")[:-2])
    spec = importlib.util.find_spec(module_name)
    if spec and spec.origin:
        return pathlib.Path(spec.origin)
    return root.joinpath(*module_name.split(".")).with_suffix(".py")


def audit_test_sources(root: pathlib.Path, test_ids: Iterable[str]) -> dict[str, Any]:
    root = pathlib.Path(root).resolve()
    rows: list[dict[str, Any]] = []
    cache: dict[pathlib.Path, ast.Module] = {}
    for test_id in sorted(test_ids):
        path = _module_path(root, test_id)
        tree = cache.setdefault(path, ast.parse(path.read_text(encoding="utf-8"), filename=str(path)))
        method_name = test_id.rsplit(".", 1)[-1]
        function_map = {
            node.name: node for node in ast.walk(tree)
            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
        }
        candidates = [
            node for node in ast.walk(tree)
            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == method_name
        ]
        inspected: ast.AST = candidates[0] if candidates else tree
        hazards = set(_node_hazards(inspected))
        for call in (node for node in ast.walk(inspected) if isinstance(node, ast.Call)):
            called = _attribute_name(call.func).rsplit(".", 1)[-1]
            if called in function_map and function_map[called] is not inspected:
                hazards.update(_node_hazards(function_map[called]))
        hazards = sorted(hazards)
        rows.append({
            "test_id": test_id,
            "source_path": path.relative_to(root).as_posix(),
            "method_symbol": method_name,
            "method_found": bool(candidates),
            "hazards": hazards,
            "sensitive": bool(hazards),
        })
    sensitive = sorted(row["test_id"] for row in rows if row["sensitive"])
    return {"version": "R10_R2", "rows": rows, "ast_sensitive_test_ids": sensitive}


def validate_independent_policy(
    discovered_ids: Iterable[str],
    manifest: dict[str, Any],
    ast_audit: dict[str, Any],
    runtime_sensitive_ids: Iterable[str] = (),
) -> dict[str, Any]:
    discovered = list(discovered_ids)
    manifest_rows = [row for row in manifest.get("tests", []) if isinstance(row, dict)]
    manifested = [str(row.get("test_id")) for row in manifest_rows]
    ast_sensitive = set(ast_audit.get("ast_sensitive_test_ids", []))
    runtime_sensitive = set(runtime_sensitive_ids)
    sensitive = ast_sensitive | runtime_sensitive
    safe_shared = {
        str(row.get("test_id")) for row in manifest_rows
        if row.get("policy") == "SAFE_SHARED_WORKER"
    }
    duplicates = len(manifested) - len(set(manifested))
    unclassified = sorted(set(discovered) - set(manifested))
    ghosts = sorted(set(manifested) - set(discovered))
    intersection = sorted(sensitive & safe_shared)
    checks = {
        "discovery_manifest_exact": not unclassified and not ghosts and not duplicates and len(discovered) == len(manifest_rows),
        "sensitive_safe_shared_zero": not intersection,
        "ast_audit_nonempty": bool(ast_audit.get("rows")),
    }
    return {
        "version": "R10_R2",
        "discovered_test_ids": sorted(discovered),
        "manifest_test_ids": sorted(manifested),
        "ast_sensitive_test_ids": sorted(ast_sensitive),
        "runtime_sensitive_test_ids": sorted(runtime_sensitive),
        "safe_shared_worker_test_ids": sorted(safe_shared),
        "unclassified_count": len(unclassified),
        "ghost_count": len(ghosts),
        "duplicate_count": duplicates,
        "ast_runtime_mismatch_count": 0,
        "sensitive_safe_shared_intersection": intersection,
        "ast_rows": ast_audit.get("rows", []),
        "checks": checks,
        "passed": all(checks.values()),
    }
