"""Pre-submission validator for the Fake Gang Detection OpenEnv environment.

Checks all submission requirements and prints pass/fail for each.

Exits 0 if all checks pass, 1 if any fail.

Usage:

python validate.py # server must be running on :8000

python validate.py --url http://host:8001

python validate.py --local # skip HTTP checks, test locally

"""

from __future__ import annotations

import argparse

import importlib

import json

import sys

import time

import urllib.error

import urllib.request

from pathlib import Path

from typing import Any, Dict, List, Optional, Tuple

_ROOT = Path(__file__).parent

sys.path.insert(0, str(_ROOT))

sys.path.insert(0, str(_ROOT / "server"))

# ---------------------------------------------------------------------------

# Helpers

# ---------------------------------------------------------------------------

_PASS = "[PASS]"

_FAIL = "[FAIL]"

_results: List[Tuple[bool, str]] = []

def check(name: str, ok: bool, detail: str = "") -> bool:

tag = _PASS if ok else _FAIL

line = f" {tag} {name}"

if detail and not ok:

line += f"\n → {detail}"

print(line)

_results.append((ok, name))

return ok

def _get(url: str) -> Tuple[Optional[dict], Optional[str]]:

try:

with urllib.request.urlopen(url, timeout=10) as r:

return json.loads(r.read()), None

except Exception as exc:

return None, str(exc)

def _post(url: str, body: Any = None) -> Tuple[Optional[dict], Optional[str]]:

try:

data = json.dumps(body or {}).encode()

req = urllib.request.Request(

url, data=data, headers={"Content-Type": "application/json"}, method="POST"

)

with urllib.request.urlopen(req, timeout=30) as r:

return json.loads(r.read()), None

except urllib.error.HTTPError as exc:

try:

body_bytes = exc.read()

return None, f"HTTP {exc.code}: {body_bytes.decode()}"

except Exception:

return None, f"HTTP {exc.code}"

except Exception as exc:

return None, str(exc)

# ---------------------------------------------------------------------------

# HTTP checks

# ---------------------------------------------------------------------------

def run_http_checks(base_url: str) -> None:

print(f"\nHTTP checks against {base_url}\n")

# /health

data, err = _get(f"{base_url}/health")

check("/health reachable", data is not None and data.get("status") == "healthy", err or "")

# /tasks — must have action_schema and 3 tasks

data, err = _get(f"{base_url}/tasks")

if check("/tasks reachable", data is not None, err or ""):

has_schema = isinstance(data.get("action_schema"), dict)

has_3_tasks = isinstance(data.get("tasks"), list) and len(data["tasks"]) == 3

has_score_range = "score_range" in data

check("/tasks has action_schema", has_schema, str(data))

check("/tasks has 3 tasks", has_3_tasks, str(data))

check("/tasks has score_range", has_score_range, str(data))

# /reset for each task

for task in ["easy", "medium", "hard"]:

data, err = _post(f"{base_url}/reset", {"task": task, "seed": 0})

check(f"/reset task={task}", data is not None and "observation" in data, err or "")

# /step — INSPECT, FLAG, SUBMIT cycle

_post(f"{base_url}/reset", {"task": "easy", "seed": 0})

obs_resp, err = _post(f"{base_url}/step", {"action_type": "inspect",

"account_id": "acc_0000"})

check("/step INSPECT", obs_resp is not None, err or "")

# Get a visible account ID from the observation to flag

acc_to_flag = None

if obs_resp:

vis_ids = obs_resp.get("observation", {}).get("visible_account_ids", [])

if vis_ids:

acc_to_flag = vis_ids[0]

if acc_to_flag:

flag_resp, err = _post(f"{base_url}/step", {"action_type": "flag",

"account_id": acc_to_flag})

check("/step FLAG", flag_resp is not None, err or "")

sub_resp, err = _post(f"{base_url}/step", {"action_type": "submit"})

check("/step SUBMIT", sub_resp is not None and sub_resp.get("done") is True, err or "")

# /grader — must return float in [0, 1]

data, err = _get(f"{base_url}/grader")

if check("/grader reachable", data is not None, err or ""):

score = data.get("score")

check("/grader returns [0,1] float",

isinstance(score, (int, float)) and 0.0 <= score <= 1.0,

f"score={score}")

# /baseline — must return 3 task scores in [0, 1]

data, err = _post(f"{base_url}/baseline")

if check("/baseline reachable", data is not None, err or ""):

scores = data.get("scores", {})

all_valid = (

set(scores.keys()) == {"easy", "medium", "hard"}

and all(isinstance(v, (int, float)) and 0.0 <= v <= 1.0

for v in scores.values())

)

check("/baseline returns 3 valid scores", all_valid,

f"got: {scores}")

# ---------------------------------------------------------------------------

# Local checks (no server needed)

# ---------------------------------------------------------------------------

def run_local_checks() -> None:

print("\nLocal checks\n")

# scoring.py importable and correct

try:

from scoring import ( # type: ignore[import]

compute_fake_risk, compute_hub_legitimacy, grader_score

)

gang_risk = compute_fake_risk(0.75, 0.65, 0.85, 0.10)

hub = compute_hub_legitimacy(2_000_000, 200, 2000, 0.05)

celeb_risk = compute_fake_risk(0.02, 0.02, 0.10, hub)

# Perfect score: 10 TP, 0 FP, 0 FN, 0 steps used → efficiency=1.0 → score=1.0

perfect = grader_score(10, 0, 0, 0, 30)

ok = (gang_risk >= 0.60 and celeb_risk < 0.20 and perfect == 1.0)

check("scoring.py math correct", ok,

f"gang_risk={gang_risk} celeb_risk={celeb_risk} perfect={perfect}")

except Exception as exc:

check("scoring.py importable", False, str(exc))

# models.py has AccountStatus + new fields

try:

from models import AccountStatus, AccountProfile, FakeGangObservation # type: ignore[import]

p = AccountProfile(

account_id="acc_0001", follower_count=100, following_count=50,

post_count=10, avg_post_hour=14.0, photo_reuse_score=0.8,

bio_template_score=0.7, account_age_days=60,

)

check("models.py AccountProfile has fake_risk_score",

hasattr(p, "fake_risk_score"), "")

check("models.py FakeGangObservation has suspect_ids",

hasattr(FakeGangObservation(), "suspect_ids"), "")

check("models.py AccountStatus enum exists",

AccountStatus.SUSPECT == "suspect", "")

except Exception as exc:

check("models.py new fields", False, str(exc))

# environment.py runs episode + status cascade

try:

from environment import FakeGangEnvironment # type: ignore[import]

from models import FakeGangAction, ActionType # type: ignore[import]

env = FakeGangEnvironment()

obs = env.reset(task="easy", seed=0)

ep_path = _ROOT / "episodes" / "easy_000.json"

if ep_path.exists():

gang_id = json.loads(ep_path.read_text())["gang_member_ids"][0]

obs = env.step(FakeGangAction(action_type=ActionType.INSPECT, account_id=gang_id))

obs = env.step(FakeGangAction(action_type=ActionType.FLAG, account_id=gang_id))

cascade_ok = len(obs.suspect_ids) > 0

check("environment.py SUSPECT cascade works", cascade_ok,

f"suspect_ids={obs.suspect_ids[:3]}")

p_flagged = next((p for p in obs.visible_accounts if p.account_id == gang_id), None)

check("environment.py fake_risk_score computed",

p_flagged is not None and p_flagged.fake_risk_score > 0, "")

else:

check("episode file exists for cascade test", False,

f"run python server/generator.py first")

except Exception as exc:

check("environment.py status cascade", False, str(exc))

# inference.py importable + runs one episode locally

try:

from inference import run_rule_based_episode # type: ignore[import]

from environment import FakeGangEnvironment # type: ignore[import]

env2 = FakeGangEnvironment()

score = run_rule_based_episode(env2, task="easy", seed=1)

check("inference.py runs locally",

isinstance(score, float) and 0.0 <= score <= 1.0,

f"score={score}")

except Exception as exc:

check("inference.py importable", False, str(exc))

# Episodes have new features

ep_path = _ROOT / "episodes" / "easy_000.json"

if ep_path.exists():

ep = json.loads(ep_path.read_text())

accounts = ep["network"]["accounts"]

first = accounts[0]["features"]

has_features = "comment_repeat_score" in first and "shared_ip_count" in first

check("episodes have new features (comment_repeat_score, shared_ip_count)",

has_features, f"keys: {list(first.keys())}")

has_celebs = "celeb_ids" in ep

check("episodes have celeb_ids field", has_celebs, "")

else:

check("episodes directory has files", False, "run python server/generator.py")

# ---------------------------------------------------------------------------

# Main

# ---------------------------------------------------------------------------

if __name__ == "__main__":

parser = argparse.ArgumentParser()

parser.add_argument("--url", default="http://localhost:8000")

parser.add_argument("--local", action="store_true",

help="Run local checks only (no server needed)")

args = parser.parse_args()

run_local_checks()

if not args.local:

run_http_checks(args.url)

total = len(_results)

passed = sum(1 for ok, _ in _results if ok)

failed = total - passed

print(f"\n{'='*50}")

print(f"Results: {passed}/{total} passed", end="")

if failed:

print(f" ({failed} FAILED)")

failed_names = [name for ok, name in _results if not ok]

for name in failed_names:

print(f" - {name}")

print()

sys.exit(1)

else:

print(" — all OK")

print()

sys.exit(0)