"""harbor.results_refresh: when the whole-program check and the emulator run are rerun, and how their results land.
The subprocess is replaced by a fake that writes what the real tools would."""
import json
import os
import subprocess
import sys
import unittest
from unittest import mock

from helpers import HarborCase
from harbor import results_refresh as rr

HOUR = 3600
T0 = 1_790_000_000.0


class Decisions(unittest.TestCase):
    """due() and choose() are pure: the schedule without any files or processes."""
    cfg = {"results_refresh": True, "match_every_minutes": 60, "behaviour_every_hours": 12, "behaviour_window": "01-06"}

    def state(self, **jobs):
        return {"jobs": jobs}

    def test_window(self):
        self.assertEqual(rr.window("01-06"), (1, 6))
        self.assertIsNone(rr.window(""))
        self.assertEqual([h for h in range(24) if rr.in_window((1, 6), h)], [1, 2, 3, 4, 5])
        self.assertEqual([h for h in range(24) if rr.in_window((22, 2), h)], [0, 1, 22, 23])
        self.assertTrue(all(rr.in_window(None, h) for h in range(24)))
        with self.assertRaises(ValueError):
            rr.window("1-30")

    def test_off_unless_enabled_but_a_request_always_runs(self):
        off = dict(self.cfg, results_refresh=False)
        self.assertEqual(rr.due(off, self.state(), "match", "abc", T0, 3), (False, "results_refresh is off"))
        self.assertEqual(rr.due(off, self.state(match={"head": "abc", "started_ts": T0}), "match", "abc", T0, 12,
                                requested={"match"}), (True, "requested"))

    def test_match_when_head_moved_at_most_hourly(self):
        s = self.state(match={"head": "old", "started_ts": T0})
        self.assertFalse(rr.due(self.cfg, self.state(match={"head": "abc", "started_ts": T0 - 9 * HOUR}),
                                "match", "abc", T0, 12)[0])                       # nothing new on main
        self.assertFalse(rr.due(self.cfg, s, "match", "new", T0 + 59 * 60, 12)[0])     # too soon
        self.assertEqual(rr.due(self.cfg, s, "match", "new", T0 + 60 * 60, 12), (True, "main moved"))
        self.assertTrue(rr.due(self.cfg, self.state(), "match", "new", T0, 12)[0])  # never ran: due at once
        self.assertTrue(rr.due(dict(self.cfg, match_every_minutes=10), s, "match", "new", T0 + 600, 12)[0])

    def test_behaviour_twelve_hourly_inside_its_window(self):
        s = self.state(behaviour={"head": "old", "started_ts": T0})
        self.assertFalse(rr.due(self.cfg, s, "behaviour", "new", T0 + 11 * HOUR, 3)[0])
        self.assertFalse(rr.due(self.cfg, s, "behaviour", "new", T0 + 13 * HOUR, 12)[0])   # outside 01-06
        self.assertTrue(rr.due(self.cfg, s, "behaviour", "new", T0 + 13 * HOUR, 1)[0])
        self.assertFalse(rr.due(self.cfg, s, "behaviour", "new", T0 + 13 * HOUR, 6)[0])    # the window ends at 06:00
        self.assertTrue(rr.due(dict(self.cfg, behaviour_window=""), s, "behaviour", "new", T0 + 13 * HOUR, 15)[0])
        self.assertFalse(rr.due(self.cfg, self.state(behaviour={"head": "new", "started_ts": T0 - 99 * HOUR}),
                                "behaviour", "new", T0, 3)[0])

    def test_choose_one_job_match_first(self):
        self.assertEqual(rr.choose(self.cfg, self.state(), "new", T0, 3)[0], "match")
        s = self.state(match={"head": "new", "started_ts": T0})
        self.assertEqual(rr.choose(self.cfg, s, "new", T0 + 60, 3)[0], "behaviour")
        job, why = rr.choose(self.cfg, s, "new", T0 + 60, 12)
        self.assertIsNone(job)
        self.assertEqual(set(why), {"match", "behaviour"})
        self.assertEqual(rr.choose(self.cfg, self.state(), "new", T0, 3, available=["behaviour"])[0], "behaviour")


class FakeProc:
    """Stands in for subprocess.Popen: writes the --json file (and a printout) the way ltcg.py / diffrun.py do."""
    def __init__(self, argv, cwd, stdout, stderr, rows=None, pid=None):
        self.argv, self.cwd, self.pid = argv, cwd, pid or os.getpid()
        out = argv[argv.index("--json") + 1]
        if rows is not None:
            with open(out, "w", encoding="utf-8") as f:
                json.dump(rows, f)
        stdout.write(b"0x401000 ResetPlayer: same\n1 functions: 1 same in 1s\n")

    def wait(self, timeout=None):
        return 0


class Refresher(HarborCase):
    def setUp(self):
        super().setUp()
        d = self.dir
        (d / "ltcg").mkdir()
        (d / "ltcg" / "ltcg.py").write_text("", encoding="utf-8")
        (d / "diffrun").mkdir()
        (d / "temp").mkdir()
        self.match_out, self.behaviour_out = d / "check.json", d / "diffrun" / "readable.json"
        self.h.project.reverse = {"ltcg": str(d / "ltcg" / "ltcg.py"), "match_results": str(self.match_out),
                                  "diffrun_command": ["python.exe", "-B", str(d / "diffrun" / "diffrun.py")],
                                  "diffrun_results": str(self.behaviour_out), "debugger_temp": str(d / "temp")}
        self.h.project.logbook = {"dir": str(d / "logbook"), "decomp": str(d / "decomp"), "results_refresh": True}
        self.head = "a" * 40
        self.calls = []
        self.real_spawn = rr._spawn
        self.rows = {"match": [{"address": "0x401000", "status": "match"}, {"address": "0x401100", "status": "differs"}],
                     "behaviour": [{"address": "0x401000", "verdict": "same"}]}
        patches = [mock.patch.object(rr, "_head", lambda project: self.head),
                   mock.patch.object(rr, "_spawn", self.spawn)]
        for p in patches:
            p.start()
            self.addCleanup(p.stop)

    def spawn(self, argv, cwd, stdout, stderr):
        job = "match" if "check" in argv else "behaviour"
        self.calls.append((job, argv, cwd))
        return FakeProc(argv, cwd, stdout, stderr, rows=self.rows.get(job))

    def state(self):
        return rr.load_state(self.h.project)

    def test_match_runs_when_main_moves_and_lands_atomically(self):
        out = rr.tick(self.h.project, now=T0, hour=12)
        self.assertTrue(out["ok"], out)
        job, argv, cwd = self.calls[0]
        self.assertEqual(job, "match")
        self.assertEqual(argv[1:5], ["-B", str(self.dir / "ltcg" / "ltcg.py"), "check", "--build"])
        self.assertEqual(argv[argv.index("--decomp") + 1], str(self.dir / "decomp"))
        self.assertEqual(argv[argv.index("--json") + 1], str(self.match_out) + ".partial")
        self.assertEqual(cwd, self.dir / "ltcg")
        self.assertEqual(json.loads(self.match_out.read_text(encoding="utf-8")), self.rows["match"])
        self.assertFalse((self.dir / "check.json.partial").exists())
        st = self.state()
        self.assertEqual((st["jobs"]["match"]["head"], st["jobs"]["match"]["counts"]),
                         (self.head, {"match": 1, "differs": 1}))
        self.assertNotIn("running", st)
        self.assertIn("idle", rr.tick(self.h.project, now=T0 + 10, hour=12))          # nothing new on main
        self.head = "b" * 40
        self.assertIn("idle", rr.tick(self.h.project, now=T0 + 30 * 60, hour=12))     # within the hour
        self.assertTrue(rr.tick(self.h.project, now=T0 + 61 * 60, hour=12)["ok"])
        self.assertEqual(len(self.calls), 2)

    def test_behaviour_at_night_with_its_printout(self):
        rr.tick(self.h.project, now=T0, hour=12)                                  # the match first
        self.assertIn("idle", rr.tick(self.h.project, now=T0 + 600, hour=12))    # emulator waits for its window
        out = rr.tick(self.h.project, now=T0 + 600, hour=2)
        self.assertEqual((out["job"], out["ok"]), ("behaviour", True))
        job, argv, cwd = self.calls[-1]
        self.assertEqual(argv[:3], self.h.project.reverse["diffrun_command"])
        self.assertEqual(argv[3:7], ["--readable", "--jobs", "4", "--json"])
        self.assertEqual(cwd, self.dir / "diffrun")
        self.assertEqual(json.loads(self.behaviour_out.read_text(encoding="utf-8")), self.rows["behaviour"])
        self.assertIn("1 functions: 1 same", (self.dir / "diffrun" / "readable.txt").read_text(encoding="utf-8"))
        self.assertEqual(self.state()["jobs"]["behaviour"]["counts"], {"same": 1})

    def test_a_failed_run_keeps_the_old_results_and_retries_after_the_interval(self):
        self.match_out.write_text('[{"old": 1}]', encoding="utf-8")
        self.rows["match"] = None                                              # the tool wrote nothing
        out = rr.tick(self.h.project, now=T0, hour=12)
        self.assertFalse(out["ok"])
        self.assertIn("no results were written", out["error"])
        self.assertEqual(self.match_out.read_text(encoding="utf-8"), '[{"old": 1}]')
        st = self.state()["jobs"]["match"]
        self.assertIsNone(st.get("head"))
        self.assertIn("idle", rr.tick(self.h.project, now=T0 + 600, hour=12))   # not hammered
        self.rows["match"] = [{"address": "0x401000", "status": "match"}]
        self.assertTrue(rr.tick(self.h.project, now=T0 + HOUR, hour=12)["ok"])

    def test_a_run_left_by_an_earlier_server_is_waited_for_then_adopted(self):
        state = {"jobs": {}, "running": {"job": "match", "pid": os.getpid(), "head": self.head,
                                         "started_ts": T0, "started": rr._iso(T0)}}
        rr.save_state(self.h.project, state)
        self.assertEqual(rr.tick(self.h.project, now=T0 + 60, hour=12)["busy"], "match")   # still going: no second run
        self.assertEqual(self.calls, [])
        done = subprocess.Popen([sys.executable, "-c", "pass"])
        done.wait()
        state["running"]["pid"] = done.pid                                     # it has finished since
        rr.save_state(self.h.project, state)
        (self.dir / "check.json.partial").write_text(json.dumps(self.rows["match"]), encoding="utf-8")
        out = rr.tick(self.h.project, now=T0 + 600, hour=12)
        self.assertTrue(out["adopted"]["ok"], out)
        self.assertEqual(self.calls, [])
        self.assertEqual(json.loads(self.match_out.read_text(encoding="utf-8")), self.rows["match"])
        self.assertEqual(self.state()["jobs"]["match"]["head"], self.head)
        self.assertNotIn("running", self.state())

    def test_waits_while_another_whole_program_build_of_main_holds_the_lock(self):
        lock = self.dir / "temp" / "ltcg-main.lock"
        lock.write_text(str(os.getpid()), encoding="utf-8")
        rr.request(self.h.project, "match")
        self.assertIn("waiting", rr.tick(self.h.project, now=T0, hour=12))
        self.assertEqual(self.calls, [])
        self.assertEqual(rr._requested(self.h.project), {"match"})             # kept for the next look
        lock.write_text("", encoding="utf-8")
        os.utime(lock, (T0 - 60, T0 - 60))                                     # an empty, old lock is stale
        self.assertTrue(rr.tick(self.h.project, now=T0 + 120, hour=12)["ok"])
        self.assertFalse(lock.exists())                                        # released after the run

    def test_a_request_runs_whatever_the_schedule(self):
        self.h.project.logbook["results_refresh"] = False
        self.assertEqual(rr.tick(self.h.project, now=T0, hour=12), {"idle": "results_refresh is off"})
        rr.request(self.h.project, "behaviour")
        out = rr.tick(self.h.project, now=T0, hour=12)
        self.assertEqual((out["job"], out["ok"]), ("behaviour", True))
        self.assertEqual(rr._requested(self.h.project), set())
        self.assertEqual(rr.tick(self.h.project, now=T0 + 5, hour=12), {"idle": "results_refresh is off"})

    def test_freshness_names_the_commit_until_someone_else_rewrites_the_file(self):
        rr.tick(self.h.project, now=T0, hour=12)
        fresh = rr.freshness(self.h.project)
        self.assertEqual((fresh["match"]["sha"], fresh["match"]["error"]), (self.head[:7], None))
        self.assertIsNone(fresh["behaviour"])
        self.assertTrue(fresh["enabled"])
        st = rr.load_state(self.h.project)
        st["jobs"]["match"]["results_mtime_ns"] -= 1                          # a different write than ours
        rr.save_state(self.h.project, st)
        self.assertIsNone(rr.freshness(self.h.project)["match"]["sha"])

    def test_runs_at_idle_priority(self):
        seen = {}

        class P:
            pid = 1

            def __init__(self, argv, **kw):
                seen.update(kw)

        with mock.patch.object(rr.subprocess, "Popen", P):
            self.real_spawn(["x"], self.dir, subprocess.DEVNULL, subprocess.DEVNULL)
        if os.name == "nt":
            self.assertTrue(seen["creationflags"] & subprocess.IDLE_PRIORITY_CLASS)
        self.assertEqual(seen["stdin"], subprocess.DEVNULL)


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