"""The state dump and the -l switch that reads it back.

The second column is the minute a torrent was last touched, by the wall
clock, which is what the file has recorded for years and the only clock two
runs of the tracker agree on.
"""

import os
import time
import unittest

from harness import Tracker, TrackerTestCase


class StateFileRoundTrip(TrackerTestCase):
    def dump(self):
        return self.tracker.get("/stats?mode=statedump")

    def test_a_dumped_torrent_comes_back_with_its_completed_count(self):
        info_hash = self.hash_for("state-rt")
        self.tracker.announce(info_hash, port=6001, left=0, event="completed")
        self.assertEqual(self.tracker.scrape(info_hash)[b"files"][info_hash][b"downloaded"], 1)

        path = os.path.join(self.tracker.dir, "state")
        with open(path, "wb") as handle:
            handle.write(self.dump())

        restarted = Tracker(extra_args=["-l", path])
        try:
            files = restarted.scrape(info_hash)[b"files"]
            self.assertIn(info_hash, files, "the torrent did not survive the restart")
            entry = files[info_hash]
            self.assertEqual(entry[b"downloaded"], 1, "the completed count did not survive")
            self.assertEqual(entry[b"complete"], 0, "no peers are carried over, only the torrent")
            self.assertEqual(entry[b"incomplete"], 0)
            complaints = restarted.stop()
            self.assertEqual(complaints, [], "the tracker reported:\n" + "\n".join(complaints))
        finally:
            restarted._cleanup()

    def test_a_torrent_older_than_the_tracker_would_keep_is_not_loaded(self):
        """An age past the torrent timeout would be cleaned away on the first
        sweep, so it is never built in the first place."""
        info_hash = self.hash_for("state-stale")
        path = os.path.join(self.tracker.dir, "stale")
        with open(path, "w") as handle:
            handle.write("%s:%d:7\n" % (info_hash.hex(), 24 * 60 + 1))

        restarted = Tracker(extra_args=["-l", path])
        try:
            self.assertNotIn(info_hash, restarted.scrape(info_hash)[b"files"])
            restarted.stop()
        finally:
            restarted._cleanup()


class StateFileWithNonsenseStamps(TrackerTestCase):
    """The column is read as unsigned and handed on as a signed ot_time. A
    value at or above 2^63 arrived negative, cleared the timeout check, and
    put the torrent's base in the future, where the cleaner never reaches it.
    At exactly 2^63 the subtraction itself overflowed."""

    def stamps(self):
        now = int(time.time()) // 60
        return {
            "overflow": 2**63,
            "minus-one": 2**64 - 1,
            "int64-max": 2**63 - 1,
            "in-the-future": now + 60,
            "past-timeout": now - (24 * 60 + 1),
        }

    def test_no_stamp_outside_the_timeout_builds_a_torrent(self):
        stamps = self.stamps()
        lines = ["%s:%d:7" % (self.hash_for(name).hex(), stamp) for name, stamp in stamps.items()]
        lines.append("%s:%d:7" % (self.hash_for("honest").hex(), int(time.time()) // 60 - 5))

        path = os.path.join(self.tracker.dir, "nonsense")
        with open(path, "w") as handle:
            handle.write("\n".join(lines) + "\n")

        restarted = Tracker(extra_args=["-l", path])
        try:
            for name, stamp in stamps.items():
                self.assertNotIn(self.hash_for(name), restarted.scrape(self.hash_for(name))[b"files"],
                                 "the line stamped %d was loaded" % stamp)
            self.assertIn(self.hash_for("honest"), restarted.scrape(self.hash_for("honest"))[b"files"],
                          "the state file was not read at all")
            self.assertEqual(restarted.torrent_count(), 1)
            complaints = restarted.stop()
            self.assertEqual(complaints, [], "the tracker reported:\n" + "\n".join(complaints))
        finally:
            restarted._cleanup()


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