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

Nothing exercised the load path before, and the dump's second column is now
how long ago a torrent was last touched rather than when, because the clock
counts from boot and an absolute stamp would mean nothing to the process
that reads the file.
"""

import os
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 StateFileWithNonsenseAges(TrackerTestCase):
    """The age 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."""

    AGES = {
        "overflow": 2**63,
        "wrapped": 2**63 + 60 * 24,
        "minus-one": 2**64 - 1,
        "int64-max": 2**63 - 1,
        "past-timeout": 60 * 24 + 1,
    }

    def test_no_age_outside_the_timeout_builds_a_torrent(self):
        lines = ["%s:%d:7" % (self.hash_for(name).hex(), age) for name, age in self.AGES.items()]
        lines.append("%s:5:7" % self.hash_for("honest").hex())

        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 in self.AGES:
                self.assertNotIn(self.hash_for(name), restarted.scrape(self.hash_for(name))[b"files"],
                                 "the line with age %d was loaded" % self.AGES[name])
            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()