"""Black-box user tasks in disposable copies; no import of analysis functions.

The numerical constants in the real-case check were independently calculated
from the publisher JATS with Fraction arithmetic, not copied from CLI output.
"""
import csv
import io
import json
import os
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile
import unittest

ROOT = Path(__file__).resolve().parents[1]
# -I -S disables user/site packages and Python environment path injection.
# This audit hook blocks Python socket API events; it is not an OS firewall.
LAUNCHER = r'''
import runpy, sys
sys.dont_write_bytecode = True
attempts = []
def no_sockets(event, arguments):
    if event.startswith("socket."):
        attempts.append(event)
        raise RuntimeError("QA blocked a Python socket API event: " + event)
sys.addaudithook(no_sockets)
import socket
try:
    socket.socket()
except RuntimeError:
    pass
else:
    raise AssertionError("Socket guard did not block its self-test")
attempts.clear()
sys.path.insert(0, sys.argv.pop(1))
sys.argv[0] = "ligninqc"
try:
    runpy.run_module("ligninqc", run_name="__main__")
finally:
    print("QA_SOCKET_ATTEMPTS=" + str(len(attempts)), file=sys.stderr)
'''


class CliUserTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory(prefix="ligninqc user QA ")
        self.addCleanup(self.temp.cleanup)
        self.home = Path(self.temp.name)
        self.copy = self.home / "unpacked release with spaces"
        shutil.copytree(ROOT, self.copy, ignore=shutil.ignore_patterns("__pycache__", "work-results", ".git"))

    def run_cli(self, *arguments):
        env = {"PATH": os.environ.get("PATH", ""),
               "LANG": "C.UTF-8", "LC_ALL": "C.UTF-8", "TMPDIR": str(self.home)}
        result = subprocess.run([sys.executable, "-I", "-S", "-B", "-c", LAUNCHER, str(self.copy), *map(str, arguments)],
                                cwd=self.home, env=env, text=True, capture_output=True, timeout=20)
        self.assertIn("QA_SOCKET_ATTEMPTS=0", result.stderr)
        return result

    def analyze(self, path, out, *extra):
        return self.run_cli("analyze", "--csv", path, "--x", "x", "--y", "y", "--id", "compound_id",
                            "--direction", "negative", "--x-unit", "kcal/mol", "--y-unit", "RDI", "--out", out, *extra)

    def input(self, content, name="input.csv"):
        path = self.home / name
        path.write_text(content, encoding="utf-8")
        return path

    def assert_rejected(self, result, out):
        self.assertNotEqual(result.returncode, 0, result.stdout)
        self.assertNotIn("Traceback", result.stderr)
        self.assertFalse(out.exists(), "Invalid input must be rejected before output is created")

    def test_real_cases_reproduce_without_network_or_site_packages(self):
        out = self.home / "results"
        process = self.run_cli("reproduce", "--out", out)
        self.assertEqual(process.returncode, 0, process.stderr)
        result = json.loads((out / "results.json").read_text())
        self.assertEqual(result["source_verification"]["lauberte_model_rows"], 19)
        self.assertEqual(result["source_verification"]["thermal_temperature_rows"], 13)
        d = result["lauberte2019"]["analyses"]["dpph_neutral_all"]
        self.assertEqual(d["n"], 18)
        self.assertAlmostEqual(d["r_squared"], 0.6365670629053138, places=11)
        self.assertAlmostEqual(d["leave_one_out"]["linear"]["mae"], 0.3221069636148318, places=11)
        self.assertEqual(d["ranking"]["discordant"], 23)
        self.assertTrue((out / "report.html").is_file())

    def test_custom_csv_from_real_table_matches_case(self):
        with (self.copy / "data/lauberte2019/table1.csv").open(newline="", encoding="utf-8") as stream:
            rows = list(csv.DictReader(stream))
        buf = io.StringIO()
        writer = csv.writer(buf)
        writer.writerow(["compound_id", "x", "y"])
        for row in reversed(rows):
            writer.writerow([row["compound_id"], row["bde_first_kcal_mol"], row["dpph_rdi"]])
        path = self.input(buf.getvalue(), "own real table.csv")
        out = self.home / "own result"
        process = self.analyze(path, out)
        self.assertEqual(process.returncode, 0, process.stderr)
        result = json.loads((out / "results.json").read_text())["analysis"]
        self.assertEqual(result["n"], 18)
        self.assertEqual(result["missing"], [{"id": "17", "missing": ["y"]}])
        self.assertAlmostEqual(result["r_squared"], 0.6365670629053138, places=11)
        self.assertAlmostEqual(result["leave_one_out"]["training_mean_baseline"]["mae"], 0.6327843137254902, places=11)

    def test_compare_min_250_checks_the_actual_prose_interval(self):
        out = self.home / "comparison"
        process = self.run_cli("compare", "--csv", self.copy / "data/kazachenko2022/table2.csv",
                               "--index", "temperature_c", "--left", "initial_mass_loss_pct",
                               "--right", "sulfated_mass_loss_pct", "--expected", "greater", "--min", "250",
                               "--unit", "% mass loss", "--out", out)
        self.assertEqual(process.returncode, 0, process.stderr)
        data = json.loads((out / "results.json").read_text())["comparison"]
        self.assertEqual((data["n"], data["agreeing"]), (12, 11))
        self.assertEqual([r["index"] for r in data["counterexamples"]], [250.0])
        self.assertAlmostEqual(data["counterexamples"][0]["right_minus_left"], -1.5)

    def test_bad_inputs_are_explained_before_any_output(self):
        cases = {
            "missing-column": "compound_id,x\na,1\nb,2\nc,3\n",
            "duplicate-id": "compound_id,x,y\na,1,3\na,2,2\nc,3,1\n",
            "duplicate-header": "compound_id,x,y,y\na,1,3,3\nb,2,2,2\nc,3,1,1\n",
            "wrong-fields": "compound_id,x,y\na,1,3\nb,2\nc,3,1\n",
            "constant": "compound_id,x,y\na,1,3\nb,1,2\nc,1,1\n",
            "nonfinite": "compound_id,x,y\na,1,3\nb,2,nan\nc,3,1\n",
            "infinite": "compound_id,x,y\na,1,3\nb,2,inf\nc,3,1\n",
            "too-few": "compound_id,x,y\na,1,3\nb,2,2\nc,3,NA\n",
            "malformed-value": "compound_id,x,y\na,1,3\nb,2,?\nc,3,1\n",
        }
        for name, content in cases.items():
            with self.subTest(name=name):
                path = self.input(content, name + ".csv")
                out = self.home / (name + "-output")
                self.assert_rejected(self.analyze(path, out), out)

    def test_missing_pair_is_visible_and_zero_is_a_measurement(self):
        path = self.input("compound_id,x,y\na,1,0\nb,2,2\nc,3,4\nd,4,NA\n")
        out = self.home / "missing-result"
        process = self.analyze(path, out)
        self.assertEqual(process.returncode, 0, process.stderr)
        data = json.loads((out / "results.json").read_text())["analysis"]
        self.assertEqual(data["n"], 3)
        self.assertEqual(data["points"][0]["y"], 0)
        self.assertEqual(data["missing"], [{"id": "d", "missing": ["y"]}])

    def test_empty_units_are_rejected(self):
        path = self.input("compound_id,x,y\na,1,3\nb,2,2\nc,3,1\n")
        out = self.home / "unit-result"
        self.assert_rejected(self.analyze(path, out, "--x-unit", "  "), out)

    def test_nonfinite_interval_is_rejected_before_output(self):
        for minimum in ("nan", "inf", "-inf"):
            with self.subTest(minimum=minimum):
                out = self.home / (minimum + "-range")
                result = self.run_cli("compare", "--csv", self.copy / "data/kazachenko2022/table2.csv",
                                      "--index", "temperature_c", "--left", "initial_mass_loss_pct",
                                      "--right", "sulfated_mass_loss_pct", "--expected", "greater", "--min=" + minimum,
                                      "--unit", "% mass loss", "--out", out)
                self.assert_rejected(result, out)

    def test_numerically_duplicate_indices_are_rejected(self):
        path = self.input("t,a,b\n250,7.7,6.2\n250.0,7.7,6.2\n300,14.4,14.7\n")
        out = self.home / "duplicated-index"
        result = self.run_cli("compare", "--csv", path, "--index", "t", "--left", "a", "--right", "b",
                              "--expected", "greater", "--unit", "% mass loss", "--out", out)
        self.assert_rejected(result, out)

    def test_existing_output_is_preserved_without_force(self):
        path = self.input("compound_id,x,y\na,1,3\nb,2,2\nc,3,1\n")
        out = self.home / "existing"
        out.mkdir()
        marker = out / "personal-note.txt"
        marker.write_text("preserve me", encoding="utf-8")
        result = self.analyze(path, out)
        self.assertNotEqual(result.returncode, 0)
        self.assertEqual(marker.read_text(), "preserve me")
        self.assertFalse((out / "results.json").exists())


if __name__ == "__main__":
    unittest.main()
