import os
import pathlib
import shutil
import stat
import subprocess
import tempfile
import unittest

import deliver
import scratch


def numbered_lines(count):
    return "".join("line {}\n".format(number) for number in range(1, count + 1))


def make_writable_and_retry(function, path, error):
    os.chmod(path, stat.S_IWRITE)
    function(path)


class Repositories(object):
    """Two real repositories, the second a real svn:externals of the first, plus a main and a workspace checkout.

    The workspace is workspace 01 of a scratch root, as delivering from a real one needs.
    """

    def __init__(self):
        self.root = tempfile.mkdtemp(prefix="delivertest-")
        self.scratch_root = os.path.join(self.root, "scratch")
        os.makedirs(self.scratch_root)
        self.record_path = scratch.deliveries_path(self.scratch_root, "01")
        self.modules_url = self.create_repository("modules")
        self.root_url = self.create_repository("root")
        self.seed_modules = self.checkout(self.modules_url, "seed-modules")
        self.write(self.seed_modules, "a.txt", numbered_lines(30))
        self.write(self.seed_modules, "b.txt", numbered_lines(30))
        self.write(self.seed_modules, os.path.join("dir", "c.txt"), numbered_lines(5))
        self.svn("add", "a.txt", "b.txt", "dir", cwd=self.seed_modules)
        self.svn("commit", "-m", "modules", cwd=self.seed_modules)
        seed_root = self.checkout(self.root_url, "seed-root")
        self.write(seed_root, "root.txt", numbered_lines(30))
        self.svn("add", "root.txt", cwd=seed_root)
        self.svn("propset", "svn:externals", "{} Modules".format(self.modules_url), ".", cwd=seed_root)
        self.svn("commit", "-m", "root", cwd=seed_root)
        self.main = self.checkout(self.root_url, "main")
        self.workspace = self.checkout(self.root_url, os.path.join("scratch", "01"))

    def cleanup(self):
        shutil.rmtree(self.root, onerror=make_writable_and_retry)

    def svn(self, *arguments, cwd=None):
        completed = subprocess.run(["svn"] + list(arguments) + ["--non-interactive"], cwd=cwd, capture_output=True,
                                   text=True)
        if completed.returncode != 0:
            raise RuntimeError("svn {} failed:\n{}".format(" ".join(arguments), completed.stderr))

        return completed.stdout

    def create_repository(self, name):
        path = os.path.join(self.root, name)
        subprocess.run(["svnadmin", "create", path], check=True)
        return pathlib.Path(path).as_uri()

    def checkout(self, url, name):
        path = os.path.join(self.root, name)
        self.svn("checkout", url, path)
        return path

    def write(self, directory, relative_path, text):
        path = os.path.join(directory, relative_path)
        os.makedirs(os.path.dirname(path), exist_ok=True)
        with open(path, "wb") as file:
            file.write(text.encode("utf-8"))

    def read(self, directory, relative_path):
        with open(os.path.join(directory, relative_path), "rb") as file:
            return file.read().decode("utf-8")

    def edit_in_workspace(self, relative_path, text, changelist="CL"):
        """Rewrites a file in the workspace, then puts it in a changelist."""
        self.write(self.workspace, relative_path, text)
        self.svn("changelist", changelist, os.path.join(self.workspace, relative_path))

    def status(self, directory):
        output = self.svn("status", "--ignore-externals", directory).replace(directory, "<wc>")
        return "".join(line for line in output.splitlines(keepends=True) if not line.startswith("X"))

    def status_of_main(self):
        return self.status(os.path.join(self.main, "Modules")) + self.status(self.main)

    def deliver(self, name="CL", dry_run=False):
        messages = []
        files = deliver.deliver(self.workspace, self.main, name, dry_run=dry_run, log=messages.append,
                                record_path=self.record_path)
        return files, messages

    def commit_main(self):
        """Commits everything pending in main, the way the user does after reviewing a delivery."""
        self.svn("commit", "-m", "reviewed", os.path.join(self.main, "Modules"))
        self.svn("commit", "-m", "reviewed", self.main)

    def commit_main_keeping_changelists(self):
        """Commits everything pending in main the way SmartSVN does, which leaves each file tagged with its changelist."""
        self.svn("commit", "--keep-changelists", "-m", "reviewed", os.path.join(self.main, "Modules"))
        self.svn("commit", "--keep-changelists", "-m", "reviewed", self.main)

    def add_to_workspace(self, relative_path, text, changelist="CL"):
        """Adds a new file in the workspace, then puts it in a changelist."""
        self.write(self.workspace, relative_path, text)
        self.svn("add", os.path.join(self.workspace, relative_path))
        self.svn("changelist", changelist, os.path.join(self.workspace, relative_path))

    def move_in_workspace(self, source, destination, changelist="CL", tag_source=True):
        """Moves a file in the workspace, then puts both ends of the move in a changelist."""
        self.svn("move", "--parents", os.path.join(self.workspace, source), os.path.join(self.workspace, destination))
        self.svn("changelist", changelist, os.path.join(self.workspace, destination))
        if tag_source:
            self.svn("changelist", changelist, os.path.join(self.workspace, source))

    def landed(self, name="CL", dry_run=False):
        messages = []
        files = deliver.landed(self.workspace, self.main, name, self.record_path, dry_run=dry_run,
                               log=messages.append)
        return files, messages

    def workspace_status(self):
        return self.status(os.path.join(self.workspace, "Modules")) + self.status(self.workspace)

    def record_for(self, name="CL"):
        return deliver.read_records(self.record_path).get(name)


class DeliverFixture(unittest.TestCase):
    def setUp(self):
        self.repositories = Repositories()
        self.addCleanup(self.repositories.cleanup)


class ModifiedFileTests(DeliverFixture):
    def test_DeliversAnEditToAFileInAnExternalAndOneInTheRoot(self):
        edited_module = numbered_lines(29) + "edited\n"
        edited_root = "edited\n" + numbered_lines(29)
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), edited_module)
        self.repositories.edit_in_workspace("root.txt", edited_root)
        files, messages = self.repositories.deliver()
        self.assertEqual(2, len(files))
        self.assertEqual(edited_module, self.repositories.read(self.repositories.main, "Modules/a.txt"))
        self.assertEqual(edited_root, self.repositories.read(self.repositories.main, "root.txt"))
        status = self.repositories.svn("status", "--ignore-externals", self.repositories.main, "--cl", "CL")
        self.assertIn("M       {}".format(os.path.join(self.repositories.main, "root.txt")), status)

    def test_PutsTheDeliveredFilesInTheChangelistInEachWorkingCopy(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "edited\n")
        self.repositories.deliver()
        status = self.repositories.svn("status", os.path.join(self.repositories.main, "Modules"))
        self.assertIn("--- Changelist 'CL':", status)

    def test_DeliversAFileThatQuotesTheBinaryFileWarning(self):
        edited = numbered_lines(29) + "Cannot display: file marked as a binary type.\n"
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), edited)
        self.repositories.deliver()
        self.assertEqual(edited, self.repositories.read(self.repositories.main, "Modules/a.txt"))

    def test_DeliversAPropertyChange(self):
        self.repositories.svn("propset", "svn:eol-style", "native",
                              os.path.join(self.repositories.workspace, "Modules", "a.txt"))
        self.repositories.svn("changelist", "CL", os.path.join(self.repositories.workspace, "Modules", "a.txt"))
        self.repositories.deliver()
        self.assertEqual("native\n", self.repositories.svn(
            "propget", "svn:eol-style", os.path.join(self.repositories.main, "Modules", "a.txt")).replace(
            "\r\n", "\n"))

class AddAndDeleteTests(DeliverFixture):
    def test_DeliversAnAddedFileInANewDirectory(self):
        workspace_modules = os.path.join(self.repositories.workspace, "Modules")
        self.repositories.svn("mkdir", os.path.join(workspace_modules, "newdir"))
        self.repositories.write(workspace_modules, os.path.join("newdir", "n.txt"), "new\n")
        self.repositories.svn("add", os.path.join(workspace_modules, "newdir", "n.txt"))
        self.repositories.svn("changelist", "CL", os.path.join(workspace_modules, "newdir", "n.txt"))
        self.repositories.deliver()
        main_modules = os.path.join(self.repositories.main, "Modules")
        self.assertEqual("new\n", self.repositories.read(main_modules, os.path.join("newdir", "n.txt")))
        status = self.repositories.svn("status", main_modules)
        self.assertIn("A       {}".format(os.path.join(main_modules, "newdir")), status)

    def test_DeliversADeletedFile(self):
        workspace_file = os.path.join(self.repositories.workspace, "Modules", "b.txt")
        self.repositories.svn("delete", workspace_file)
        self.repositories.svn("changelist", "CL", workspace_file)
        self.repositories.deliver()
        self.assertFalse(os.path.exists(os.path.join(self.repositories.main, "Modules", "b.txt")))
        status = self.repositories.svn("status", os.path.join(self.repositories.main, "Modules"))
        self.assertIn("D       {}".format(os.path.join(self.repositories.main, "Modules", "b.txt")), status)

class RefusalTests(DeliverFixture):
    def assert_main_untouched(self, expected_status=""):
        self.assertEqual(expected_status, self.repositories.status_of_main())

    def test_RefusesWhenTheChangelistDoesNotExist(self):
        with self.assertRaises(deliver.DeliveryRefused):
            self.repositories.deliver("NoSuchChangelist")

    def test_RefusesAFileMainHasAlreadyModified(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "mine\n")
        self.repositories.write(self.repositories.main, os.path.join("Modules", "a.txt"), "someone else's\n")
        with self.assertRaises(deliver.DeliveryRefused):
            self.repositories.deliver()
        self.assertEqual("someone else's\n", self.repositories.read(self.repositories.main, "Modules/a.txt"))

    def test_RefusesAFileInAnotherChangelistInMain(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "mine\n")
        main_file = os.path.join(self.repositories.main, "Modules", "a.txt")
        self.repositories.write(self.repositories.main, os.path.join("Modules", "a.txt"), "theirs\n")
        self.repositories.svn("changelist", "Theirs", main_file)
        with self.assertRaises(deliver.DeliveryRefused) as caught:
            self.repositories.deliver()
        self.assertIn("Theirs", str(caught.exception))
        self.assertEqual("theirs\n", self.repositories.read(self.repositories.main, "Modules/a.txt"))

    def test_RefusesABinaryFile(self):
        workspace_modules = os.path.join(self.repositories.workspace, "Modules")
        self.repositories.write(workspace_modules, "blob.dat", "binary\n")
        self.repositories.svn("add", os.path.join(workspace_modules, "blob.dat"))
        self.repositories.svn("propset", "svn:mime-type", "application/octet-stream",
                              os.path.join(workspace_modules, "blob.dat"))
        self.repositories.svn("changelist", "CL", os.path.join(workspace_modules, "blob.dat"))
        with self.assertRaises(deliver.DeliveryRefused) as caught:
            self.repositories.deliver()
        self.assertIn("binary", str(caught.exception))
        self.assertFalse(os.path.exists(os.path.join(self.repositories.main, "Modules", "blob.dat")))

    def test_RefusesAFileWithCopyHistory(self):
        workspace_modules = os.path.join(self.repositories.workspace, "Modules")
        self.repositories.svn("copy", os.path.join(workspace_modules, "a.txt"),
                              os.path.join(workspace_modules, "copy.txt"))
        self.repositories.svn("changelist", "CL", os.path.join(workspace_modules, "copy.txt"))
        with self.assertRaises(deliver.DeliveryRefused) as caught:
            self.repositories.deliver()
        self.assertIn("history", str(caught.exception))

    def test_RefusesADeletedDirectory(self):
        workspace_modules = os.path.join(self.repositories.workspace, "Modules")
        self.repositories.svn("delete", os.path.join(workspace_modules, "dir"))
        self.repositories.svn("changelist", "CL", os.path.join(workspace_modules, "dir", "c.txt"))
        with self.assertRaises(deliver.DeliveryRefused) as caught:
            self.repositories.deliver()
        self.assertIn("deleted directory", str(caught.exception))

    def test_RollsMainBackWhenThePatchDoesNotApply(self):
        seed = self.repositories.seed_modules
        upstream = numbered_lines(30).replace("line 2\n", "changed upstream\n")
        self.repositories.write(seed, "a.txt", upstream)
        self.repositories.svn("commit", "-m", "upstream", cwd=seed)
        self.repositories.svn("update", os.path.join(self.repositories.workspace, "Modules"))
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), upstream.replace("changed upstream",
                                                                                        "mine"))
        with self.assertRaises(deliver.DeliveryRefused):
            self.repositories.deliver()
        self.assertEqual(numbered_lines(30), self.repositories.read(self.repositories.main, "Modules/a.txt"))
        self.assertEqual("", self.repositories.status_of_main().strip())

class RollbackDirectoryTests(DeliverFixture):
    def test_RemovesADirectoryItCreatedWhenThePatchDoesNotApply(self):
        seed = self.repositories.seed_modules
        upstream = numbered_lines(30).replace("line 2\n", "changed upstream\n")
        self.repositories.write(seed, "a.txt", upstream)
        self.repositories.svn("commit", "-m", "upstream", cwd=seed)
        self.repositories.svn("update", os.path.join(self.repositories.workspace, "Modules"))
        workspace_modules = os.path.join(self.repositories.workspace, "Modules")
        self.repositories.svn("mkdir", os.path.join(workspace_modules, "newdir"))
        self.repositories.write(workspace_modules, os.path.join("newdir", "n.txt"), "new\n")
        self.repositories.svn("add", os.path.join(workspace_modules, "newdir", "n.txt"))
        self.repositories.svn("changelist", "CL", os.path.join(workspace_modules, "newdir", "n.txt"))
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"),
                                            upstream.replace("changed upstream", "mine"))
        with self.assertRaises(deliver.DeliveryRefused):
            self.repositories.deliver()
        self.assertFalse(os.path.exists(os.path.join(self.repositories.main, "Modules", "newdir")))
        self.assertEqual("", self.repositories.status_of_main().strip())


class VerifyTests(DeliverFixture):
    def delivered_plan(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "edited\n")
        self.repositories.deliver()
        return [plan for plan in deliver.plan_delivery(self.repositories.workspace, self.repositories.main, "CL")][0]

    def test_FindsNothingAfterAFaithfulDelivery(self):
        self.assertEqual([], deliver.verify(self.delivered_plan(), "CL"))

    def test_FindsAFileThatDiffersFromTheWorkspace(self):
        plan = self.delivered_plan()
        self.repositories.write(self.repositories.main, os.path.join("Modules", "a.txt"), "tampered\n")
        problems = deliver.verify(plan, "CL")
        self.assertEqual(1, len(problems))
        self.assertIn("differs from the workspace", problems[0])

    def test_FindsAFileThatIsNotInTheChangelist(self):
        plan = self.delivered_plan()
        self.repositories.svn("changelist", "--remove", os.path.join(self.repositories.main, "Modules", "a.txt"))
        problems = deliver.verify(plan, "CL")
        self.assertEqual(1, len(problems))
        self.assertIn("changelist 'CL'", problems[0])


class UnsupportedReasonTests(unittest.TestCase):
    def reason(self, flags):
        return deliver.unsupported_reason(deliver.StatusEntry(flags, "x", "CL"))

    def test_AcceptsAddedDeletedModifiedAndPropertyOnlyChanges(self):
        for flags in ("A      ", "D      ", "M      ", " M     ", "MM     "):
            self.assertIsNone(self.reason(flags), flags)

    def test_RefusesStatusesThatAreNotChanges(self):
        for flags in ("!      ", "~      ", "R      ", "C      ", "?      "):
            self.assertIn("has the status", self.reason(flags), flags)

    def test_RefusesCopyHistory(self):
        self.assertIn("history", self.reason("A  +   "))

    def test_RefusesASwitchedPath(self):
        self.assertIn("switched", self.reason("M   S  "))

    def test_RefusesATreeConflict(self):
        self.assertIn("tree conflict", self.reason("A     C"))


class DryRunTests(DeliverFixture):
    def test_ReportsTheFilesAndChangesNothing(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "edited\n")
        files, messages = self.repositories.deliver(dry_run=True)
        self.assertEqual(1, len(files))
        self.assertEqual("", self.repositories.status_of_main().strip())
        self.assertTrue(any("a.txt" in message for message in messages))

class RedeliveryTests(DeliverFixture):
    def test_ReplacesTheEarlierDeliveryWithTheWorkspacesLatest(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "first\n")
        self.repositories.deliver()
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "second\n")
        files, messages = self.repositories.deliver()
        self.assertEqual(numbered_lines(29) + "second\n",
                         self.repositories.read(self.repositories.main, "Modules/a.txt"))
        self.assertTrue(any("replacing an earlier delivery" in message for message in messages))

    def test_RevertsAFileNoLongerInTheChangelist(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "edited\n")
        self.repositories.edit_in_workspace(os.path.join("Modules", "b.txt"), numbered_lines(29) + "edited\n")
        self.repositories.deliver()
        self.repositories.svn("revert", os.path.join(self.repositories.workspace, "Modules", "b.txt"))
        self.repositories.deliver()
        self.assertEqual(numbered_lines(30), self.repositories.read(self.repositories.main, "Modules/b.txt"))
        self.assertNotIn("b.txt", self.repositories.status_of_main())

    def test_RemovesADirectoryThatNoLongerHoldsAnything(self):
        workspace_modules = os.path.join(self.repositories.workspace, "Modules")
        self.repositories.svn("mkdir", os.path.join(workspace_modules, "newdir"))
        self.repositories.write(workspace_modules, os.path.join("newdir", "n.txt"), "new\n")
        self.repositories.svn("add", os.path.join(workspace_modules, "newdir", "n.txt"))
        self.repositories.svn("changelist", "CL", os.path.join(workspace_modules, "newdir", "n.txt"))
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "edited\n")
        self.repositories.deliver()
        self.repositories.svn("revert", "-R", os.path.join(workspace_modules, "newdir"))
        self.repositories.deliver()
        self.assertFalse(os.path.exists(os.path.join(self.repositories.main, "Modules", "newdir")))

class LeftoverChangelistTagTests(DeliverFixture):
    def leave_a_committed_file_tagged_in_main(self, relative_path, name="Theirs"):
        """Commits a change to a file from main so that it stays tagged with the changelist, then catches the
        workspace up."""
        main_file = os.path.join(self.repositories.main, relative_path)
        self.repositories.write(self.repositories.main, relative_path, numbered_lines(29) + "theirs\n")
        self.repositories.svn("changelist", name, main_file)
        self.repositories.commit_main_keeping_changelists()
        self.repositories.svn("update", os.path.join(self.repositories.workspace, "Modules"))
        self.repositories.svn("update", self.repositories.workspace)

    def test_DeliversOverAFileWhoseOldChangelistIsAllThatRemains(self):
        self.leave_a_committed_file_tagged_in_main(os.path.join("Modules", "b.txt"))
        self.repositories.edit_in_workspace(os.path.join("Modules", "b.txt"), numbered_lines(29) + "mine\n")
        files, messages = self.repositories.deliver()
        self.assertEqual(1, len(files))
        self.assertEqual(numbered_lines(29) + "mine\n", self.repositories.read(self.repositories.main, "Modules/b.txt"))
        status = self.repositories.svn("status", os.path.join(self.repositories.main, "Modules"), "--cl", "CL")
        self.assertIn("b.txt", status)
        self.assertFalse(any("replacing" in message for message in messages))

    def test_StillRefusesAFileTheOldChangelistHoldsWithAPendingChange(self):
        main_file = os.path.join(self.repositories.main, "Modules", "b.txt")
        self.repositories.write(self.repositories.main, os.path.join("Modules", "b.txt"), "theirs, uncommitted\n")
        self.repositories.svn("changelist", "Theirs", main_file)
        self.repositories.edit_in_workspace(os.path.join("Modules", "b.txt"), numbered_lines(29) + "mine\n")
        with self.assertRaises(deliver.DeliveryRefused) as caught:
            self.repositories.deliver()
        self.assertIn("Theirs", str(caught.exception))

    def test_ADeliveryAgainClearsTheTagItLeftOnAFileThatWasCommitted(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "first\n")
        self.repositories.deliver()
        self.repositories.commit_main_keeping_changelists()
        self.repositories.svn("update", os.path.join(self.repositories.workspace, "Modules"))
        self.repositories.edit_in_workspace(os.path.join("Modules", "b.txt"), numbered_lines(29) + "second\n")
        files, messages = self.repositories.deliver()
        self.assertEqual(1, len(files))
        self.assertFalse(any("replacing" in message for message in messages))

    def test_LandedClearsTheTagFromTheFilesItLanded(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "edited\n")
        self.repositories.deliver()
        self.repositories.commit_main_keeping_changelists()
        self.repositories.landed()
        self.assertEqual("", self.repositories.status_of_main().strip())

    def test_LandedLeavesTheTagOnAFileItDidNotDeliver(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "edited\n")
        self.repositories.deliver()
        other = os.path.join(self.repositories.main, "Modules", "b.txt")
        self.repositories.write(self.repositories.main, os.path.join("Modules", "b.txt"), "pending\n")
        self.repositories.svn("changelist", "CL", other)
        self.repositories.svn("commit", "--keep-changelists", "-m", "reviewed",
                              os.path.join(self.repositories.main, "Modules", "a.txt"))
        self.repositories.landed()
        status = self.repositories.svn("status", os.path.join(self.repositories.main, "Modules"), "--cl", "CL")
        self.assertIn("b.txt", status)
        self.assertNotIn("a.txt", status)


class RollbackSafetyTests(DeliverFixture):
    def deliver_a_new_file_and_commit_it(self):
        self.repositories.add_to_workspace(os.path.join("Modules", "n.txt"), "new\n")
        self.repositories.deliver()
        self.repositories.commit_main_keeping_changelists()

    def test_RefusesToDeliverAgainANewFileThatHasBeenCommitted(self):
        self.deliver_a_new_file_and_commit_it()
        with self.assertRaises(deliver.DeliveryRefused) as caught:
            self.repositories.deliver()
        self.assertIn("--landed", str(caught.exception))
        self.assertIn("n.txt", str(caught.exception))

    def test_KeepsACommittedFileOnDiskWhenItRefusesToDeliverAgain(self):
        self.deliver_a_new_file_and_commit_it()
        with self.assertRaises(deliver.DeliveryRefused):
            self.repositories.deliver()
        self.assertEqual("new\n", self.repositories.read(self.repositories.main, "Modules/n.txt"))
        self.assertNotIn("!", self.repositories.status_of_main())

    def test_RefusesANewFileThatIsAlreadyInMainUnversioned(self):
        self.repositories.add_to_workspace(os.path.join("Modules", "n.txt"), "new\n")
        self.repositories.write(self.repositories.main, os.path.join("Modules", "n.txt"), "stray\n")
        with self.assertRaises(deliver.DeliveryRefused) as caught:
            self.repositories.deliver()
        self.assertIn("n.txt", str(caught.exception))
        self.assertEqual("stray\n", self.repositories.read(self.repositories.main, "Modules/n.txt"))

    def test_DeliversAgainANewFileStillPendingInMain(self):
        self.repositories.add_to_workspace(os.path.join("Modules", "n.txt"), "first\n")
        self.repositories.deliver()
        self.repositories.write(self.repositories.workspace, os.path.join("Modules", "n.txt"), "second\n")
        files, messages = self.repositories.deliver()
        self.assertEqual("second\n", self.repositories.read(self.repositories.main, "Modules/n.txt"))
        self.assertTrue(any("replacing an earlier delivery" in message for message in messages))

    def test_RollBackLeavesAFileThatWasAlreadyThere(self):
        self.repositories.add_to_workspace(os.path.join("Modules", "n.txt"), "new\n")
        plans = deliver.plan_delivery(self.repositories.workspace, self.repositories.main, "CL")
        main_file = os.path.join(self.repositories.main, "Modules", "n.txt")
        self.repositories.write(self.repositories.main, os.path.join("Modules", "n.txt"), "committed\n")
        existing = set([deliver.path_key(main_file)])
        deliver.roll_back(plans, [], existing)
        self.assertEqual("committed\n", self.repositories.read(self.repositories.main, "Modules/n.txt"))

    def test_RollBackRemovesAFileTheDeliveryCreated(self):
        self.repositories.add_to_workspace(os.path.join("Modules", "n.txt"), "new\n")
        plans = deliver.plan_delivery(self.repositories.workspace, self.repositories.main, "CL")
        self.repositories.write(self.repositories.main, os.path.join("Modules", "n.txt"), "created by the patch\n")
        deliver.roll_back(plans, [], set())
        self.assertFalse(os.path.exists(os.path.join(self.repositories.main, "Modules", "n.txt")))


class MoveTests(DeliverFixture):
    def moved_in_main(self):
        """Returns the (source, destination) of each move svn records in main's Modules."""
        return deliver.moved_files(os.path.join(self.repositories.main, "Modules"))

    def test_DeliversAMoveAsAMove(self):
        self.repositories.move_in_workspace(os.path.join("Modules", "a.txt"), os.path.join("Modules", "moved.txt"))
        files, messages = self.repositories.deliver()
        self.assertEqual(2, len(files))
        main_modules = os.path.join(self.repositories.main, "Modules")
        self.assertEqual([(os.path.join(main_modules, "a.txt"), os.path.join(main_modules, "moved.txt"))],
                         self.moved_in_main())
        self.assertEqual(numbered_lines(30), self.repositories.read(main_modules, "moved.txt"))
        self.assertFalse(os.path.exists(os.path.join(main_modules, "a.txt")))
        status = self.repositories.svn("status", main_modules, "--cl", "CL")
        self.assertIn("a.txt", status)
        self.assertIn("moved.txt", status)

    def test_DeliversAMoveThatAlsoEditsTheFile(self):
        self.repositories.move_in_workspace(os.path.join("Modules", "a.txt"), os.path.join("Modules", "moved.txt"))
        self.repositories.write(self.repositories.workspace, os.path.join("Modules", "moved.txt"),
                                numbered_lines(30) + "edited\n")
        self.repositories.deliver()
        main_modules = os.path.join(self.repositories.main, "Modules")
        self.assertEqual(numbered_lines(30) + "edited\n", self.repositories.read(main_modules, "moved.txt"))
        self.assertEqual(1, len(self.moved_in_main()))

    def test_DeliversAMoveIntoANewDirectory(self):
        self.repositories.svn("mkdir", os.path.join(self.repositories.workspace, "Modules", "newdir"))
        self.repositories.move_in_workspace(os.path.join("Modules", "a.txt"),
                                            os.path.join("Modules", "newdir", "a.txt"))
        self.repositories.deliver()
        main_modules = os.path.join(self.repositories.main, "Modules")
        self.assertEqual(numbered_lines(30), self.repositories.read(main_modules, os.path.join("newdir", "a.txt")))
        self.assertEqual(1, len(self.moved_in_main()))

    def test_DeliversAMoveOfAFileWithAProperty(self):
        workspace_file = os.path.join(self.repositories.workspace, "Modules", "a.txt")
        self.repositories.svn("propset", "svn:mime-type", "application/octet-stream", workspace_file)
        self.repositories.svn("commit", "-m", "binary", workspace_file)
        self.repositories.svn("update", self.repositories.main)
        self.repositories.move_in_workspace(os.path.join("Modules", "a.txt"), os.path.join("Modules", "moved.txt"))
        self.repositories.deliver()
        self.assertEqual(1, len(self.moved_in_main()))
        self.assertEqual("application/octet-stream", self.repositories.svn(
            "propget", "svn:mime-type", os.path.join(self.repositories.main, "Modules", "moved.txt")).strip())

    def test_MergesAMoveOverAnEarlierCommitToTheSource(self):
        upstream = numbered_lines(30).replace("line 2\n", "changed upstream\n")
        self.repositories.write(self.repositories.seed_modules, "a.txt", upstream)
        self.repositories.svn("commit", "-m", "upstream", cwd=self.repositories.seed_modules)
        self.repositories.svn("update", os.path.join(self.repositories.main, "Modules"))
        self.repositories.move_in_workspace(os.path.join("Modules", "a.txt"), os.path.join("Modules", "moved.txt"))
        self.repositories.write(self.repositories.workspace, os.path.join("Modules", "moved.txt"),
                                numbered_lines(30) + "edited\n")
        files, messages = self.repositories.deliver()
        self.assertEqual(upstream + "edited\n",
                         self.repositories.read(os.path.join(self.repositories.main, "Modules"), "moved.txt"))
        self.assertTrue(any("different base" in message for message in messages))

    def test_ReplacesAnEarlierDeliveryOfAMove(self):
        self.repositories.move_in_workspace(os.path.join("Modules", "a.txt"), os.path.join("Modules", "moved.txt"))
        self.repositories.deliver()
        self.repositories.write(self.repositories.workspace, os.path.join("Modules", "moved.txt"),
                                numbered_lines(30) + "second\n")
        files, messages = self.repositories.deliver()
        self.assertEqual(2, len(files))
        self.assertTrue(any("replacing an earlier delivery" in message for message in messages))
        main_modules = os.path.join(self.repositories.main, "Modules")
        self.assertEqual(numbered_lines(30) + "second\n", self.repositories.read(main_modules, "moved.txt"))
        self.assertEqual(1, len(self.moved_in_main()))

    def test_RefusesAMoveWhoseSourceMainHasModified(self):
        self.repositories.move_in_workspace(os.path.join("Modules", "a.txt"), os.path.join("Modules", "moved.txt"))
        self.repositories.write(self.repositories.main, os.path.join("Modules", "a.txt"), "someone else's\n")
        with self.assertRaises(deliver.DeliveryRefused):
            self.repositories.deliver()
        self.assertEqual("someone else's\n", self.repositories.read(self.repositories.main, "Modules/a.txt"))
        self.assertFalse(os.path.exists(os.path.join(self.repositories.main, "Modules", "moved.txt")))

    def test_RefusesAMoveWhoseSourceIsOutsideTheChangelist(self):
        self.repositories.move_in_workspace(os.path.join("Modules", "a.txt"), os.path.join("Modules", "moved.txt"),
                                            tag_source=False)
        with self.assertRaises(deliver.DeliveryRefused) as caught:
            self.repositories.deliver()
        self.assertIn("history", str(caught.exception))

    def test_RollsMainBackWhenAMovedFilesEditDoesNotApply(self):
        upstream = numbered_lines(30).replace("line 2\n", "changed upstream\n")
        self.repositories.write(self.repositories.seed_modules, "a.txt", upstream)
        self.repositories.svn("commit", "-m", "upstream", cwd=self.repositories.seed_modules)
        self.repositories.svn("update", os.path.join(self.repositories.main, "Modules"))
        self.repositories.move_in_workspace(os.path.join("Modules", "a.txt"), os.path.join("Modules", "moved.txt"))
        self.repositories.write(self.repositories.workspace, os.path.join("Modules", "moved.txt"),
                                numbered_lines(30).replace("line 2\n", "mine\n"))
        with self.assertRaises(deliver.DeliveryRefused):
            self.repositories.deliver()
        self.assertEqual(upstream, self.repositories.read(self.repositories.main, "Modules/a.txt"))
        self.assertFalse(os.path.exists(os.path.join(self.repositories.main, "Modules", "moved.txt")))
        self.assertEqual("", self.repositories.status_of_main().strip())

    def test_KeepsTheEarlierMoveWhenAReplacementIsRefused(self):
        self.repositories.move_in_workspace(os.path.join("Modules", "a.txt"), os.path.join("Modules", "moved.txt"))
        self.repositories.deliver()
        main_modules = os.path.join(self.repositories.main, "Modules")
        self.repositories.write(main_modules, "unrelated.txt", "stray\n")
        self.repositories.svn("add", os.path.join(main_modules, "unrelated.txt"))
        self.repositories.svn("move", os.path.join(self.repositories.workspace, "Modules", "b.txt"),
                              os.path.join(self.repositories.workspace, "Modules", "unrelated.txt"))
        self.repositories.svn("changelist", "CL", os.path.join(self.repositories.workspace, "Modules", "b.txt"),
                              os.path.join(self.repositories.workspace, "Modules", "unrelated.txt"))
        with self.assertRaises(deliver.DeliveryRefused):
            self.repositories.deliver()
        self.assertEqual("stray\n", self.repositories.read(main_modules, "unrelated.txt"))
        self.assertEqual(1, len(self.moved_in_main()))

    def test_VerifyReportsACopyAndDeleteThatIsNotAMove(self):
        self.repositories.move_in_workspace(os.path.join("Modules", "a.txt"), os.path.join("Modules", "moved.txt"))
        self.repositories.deliver()
        main_modules = os.path.join(self.repositories.main, "Modules")
        self.repositories.svn("revert", os.path.join(main_modules, "moved.txt"))
        self.repositories.svn("revert", os.path.join(main_modules, "a.txt"))
        self.repositories.svn("delete", os.path.join(main_modules, "a.txt"))
        self.repositories.write(main_modules, "moved.txt", numbered_lines(30))
        self.repositories.svn("add", os.path.join(main_modules, "moved.txt"))
        self.repositories.svn("changelist", "CL", os.path.join(main_modules, "a.txt"),
                              os.path.join(main_modules, "moved.txt"))
        plan = deliver.plan_delivery(self.repositories.workspace, self.repositories.main, "CL")[0]
        problems = deliver.verify(plan, "CL")
        self.assertEqual(1, len(problems))
        self.assertIn("moved.txt", problems[0])
        self.assertIn("move", problems[0])

    def test_DryRunListsBothEndsAndChangesNothing(self):
        self.repositories.move_in_workspace(os.path.join("Modules", "a.txt"), os.path.join("Modules", "moved.txt"))
        files, messages = self.repositories.deliver(dry_run=True)
        self.assertEqual(2, len(files))
        self.assertEqual("", self.repositories.status_of_main().strip())


class MoveLandedTests(DeliverFixture):
    def test_DropsTheWorkspacesCopyOfAMove(self):
        self.repositories.move_in_workspace(os.path.join("Modules", "a.txt"), os.path.join("Modules", "moved.txt"))
        self.repositories.deliver()
        self.repositories.commit_main()
        files, messages = self.repositories.landed()
        self.assertEqual(2, len(files))
        workspace_modules = os.path.join(self.repositories.workspace, "Modules")
        self.assertEqual(numbered_lines(30), self.repositories.read(workspace_modules, "moved.txt"))
        self.assertFalse(os.path.exists(os.path.join(workspace_modules, "a.txt")))
        self.assertEqual("", self.repositories.workspace_status().strip())

    def test_DropsAMoveThatAlsoEditsTheFile(self):
        self.repositories.move_in_workspace(os.path.join("Modules", "a.txt"), os.path.join("Modules", "moved.txt"))
        self.repositories.write(self.repositories.workspace, os.path.join("Modules", "moved.txt"),
                                numbered_lines(30) + "edited\n")
        self.repositories.deliver()
        self.repositories.commit_main()
        self.repositories.landed()
        workspace_modules = os.path.join(self.repositories.workspace, "Modules")
        self.assertEqual(numbered_lines(30) + "edited\n", self.repositories.read(workspace_modules, "moved.txt"))
        self.assertEqual("", self.repositories.workspace_status().strip())

    def test_RefusesWhileMainHasNotCommittedTheMove(self):
        self.repositories.move_in_workspace(os.path.join("Modules", "a.txt"), os.path.join("Modules", "moved.txt"))
        self.repositories.deliver()
        with self.assertRaises(deliver.DeliveryRefused) as caught:
            self.repositories.landed()
        self.assertIn("hasn't been committed", str(caught.exception))
        self.assertTrue(os.path.exists(os.path.join(self.repositories.workspace, "Modules", "moved.txt")))

    def test_RefusesAMoveEditedAfterItWasDelivered(self):
        self.repositories.move_in_workspace(os.path.join("Modules", "a.txt"), os.path.join("Modules", "moved.txt"))
        self.repositories.deliver()
        self.repositories.commit_main()
        self.repositories.write(self.repositories.workspace, os.path.join("Modules", "moved.txt"), "later\n")
        with self.assertRaises(deliver.DeliveryRefused) as caught:
            self.repositories.landed()
        self.assertIn("edited after it was delivered", str(caught.exception))


class DifferentRevisionTests(DeliverFixture):
    def test_WarnsWhenWorkspaceAndMainAreAtDifferentRevisions(self):
        seed = self.repositories.seed_modules
        self.repositories.write(seed, "b.txt", numbered_lines(29) + "upstream\n")
        self.repositories.svn("commit", "-m", "upstream", cwd=seed)
        self.repositories.svn("update", os.path.join(self.repositories.workspace, "Modules"))
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "edited\n")
        files, messages = self.repositories.deliver()
        self.assertTrue(any("different base" in message for message in messages))
        self.assertEqual(numbered_lines(29) + "edited\n",
                         self.repositories.read(self.repositories.main, "Modules/a.txt"))


class RecordTests(DeliverFixture):
    def test_RecordsEachDeliveredFileWithItsDigestsAndBaseRevision(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "edited\n")
        self.repositories.deliver()
        record = self.repositories.record_for()
        self.assertEqual(1, len(record))
        workspace_file = os.path.join(self.repositories.workspace, "Modules", "a.txt")
        self.assertEqual("Modules/a.txt", record[0]["path"])
        self.assertEqual("M ", record[0]["flags"])
        self.assertEqual(deliver.file_digest(workspace_file), record[0]["digest"])
        self.assertEqual(deliver.properties_digest(workspace_file), record[0]["properties_digest"])
        self.assertEqual(1, record[0]["base_revision"])

    def test_RecordsAnAddedFileWithNoBaseRevision(self):
        workspace_modules = os.path.join(self.repositories.workspace, "Modules")
        self.repositories.write(workspace_modules, "n.txt", "new\n")
        self.repositories.svn("add", os.path.join(workspace_modules, "n.txt"))
        self.repositories.svn("changelist", "CL", os.path.join(workspace_modules, "n.txt"))
        self.repositories.deliver()
        record = self.repositories.record_for()
        self.assertEqual("A ", record[0]["flags"])
        self.assertIsNotNone(record[0]["digest"])
        self.assertIsNone(record[0]["base_revision"])

    def test_RecordsADeletedFileWithNoDigests(self):
        workspace_file = os.path.join(self.repositories.workspace, "Modules", "b.txt")
        self.repositories.svn("delete", workspace_file)
        self.repositories.svn("changelist", "CL", workspace_file)
        self.repositories.deliver()
        record = self.repositories.record_for()
        self.assertEqual("D ", record[0]["flags"])
        self.assertIsNone(record[0]["digest"])
        self.assertIsNone(record[0]["properties_digest"])
        self.assertIsNone(record[0]["base_revision"])

    def test_KeepsEachChangelistsRecordSeparately(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "one\n")
        self.repositories.edit_in_workspace(os.path.join("Modules", "b.txt"), numbered_lines(29) + "two\n", "Other")
        self.repositories.deliver()
        self.repositories.deliver("Other")
        self.assertEqual(["Modules/a.txt"], [item["path"] for item in self.repositories.record_for("CL")])
        self.assertEqual(["Modules/b.txt"], [item["path"] for item in self.repositories.record_for("Other")])

    def test_ADeliveryAgainReplacesTheRecord(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "first\n")
        self.repositories.deliver()
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "second\n")
        self.repositories.deliver()
        self.assertEqual(deliver.file_digest(os.path.join(self.repositories.workspace, "Modules", "a.txt")),
                         self.repositories.record_for()[0]["digest"])

    def test_ADryRunRecordsNothing(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "edited\n")
        self.repositories.deliver(dry_run=True)
        self.assertFalse(os.path.exists(self.repositories.record_path))

    def test_ARefusedDeliveryRecordsNothing(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "mine\n")
        self.repositories.write(self.repositories.main, os.path.join("Modules", "a.txt"), "someone else's\n")
        with self.assertRaises(deliver.DeliveryRefused):
            self.repositories.deliver()
        self.assertFalse(os.path.exists(self.repositories.record_path))

    def test_ADeliveryWithoutARecordPathRecordsNothing(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "edited\n")
        deliver.deliver(self.repositories.workspace, self.repositories.main, "CL", log=lambda message: None)
        self.assertFalse(os.path.exists(self.repositories.record_path))


class LandedTests(DeliverFixture):
    def edit_deliver_and_commit(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "edited\n")
        self.repositories.deliver()
        self.repositories.commit_main()

    def test_DropsTheWorkspacesCopyOfAnEditedFile(self):
        self.edit_deliver_and_commit()
        files, messages = self.repositories.landed()
        self.assertEqual(1, len(files))
        self.assertEqual(numbered_lines(29) + "edited\n",
                         self.repositories.read(self.repositories.workspace, "Modules/a.txt"))
        self.assertEqual("", self.repositories.workspace_status().strip())

    def test_TakesTheEditsMadeInMainBeforeCommitting(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "edited\n")
        self.repositories.deliver()
        self.repositories.write(self.repositories.main, os.path.join("Modules", "a.txt"), "changed in main\n")
        self.repositories.commit_main()
        self.repositories.landed()
        self.assertEqual("changed in main\n", self.repositories.read(self.repositories.workspace, "Modules/a.txt"))
        self.assertEqual("", self.repositories.workspace_status().strip())

    def test_UpdatesTheWorkspaceToTheCommit(self):
        self.edit_deliver_and_commit()
        workspace_modules = os.path.join(self.repositories.workspace, "Modules")
        revision_before = deliver.working_copy_revision(workspace_modules)
        self.repositories.landed()
        self.assertEqual(revision_before + 1, deliver.working_copy_revision(workspace_modules))

    def test_DropsAnAddedFileAndTheDirectoryItWasAddedIn(self):
        workspace_modules = os.path.join(self.repositories.workspace, "Modules")
        self.repositories.svn("mkdir", os.path.join(workspace_modules, "newdir"))
        self.repositories.write(workspace_modules, os.path.join("newdir", "n.txt"), "new\n")
        self.repositories.svn("add", os.path.join(workspace_modules, "newdir", "n.txt"))
        self.repositories.svn("changelist", "CL", os.path.join(workspace_modules, "newdir", "n.txt"))
        self.repositories.deliver()
        self.repositories.commit_main()
        self.repositories.landed()
        self.assertEqual("new\n", self.repositories.read(workspace_modules, os.path.join("newdir", "n.txt")))
        self.assertEqual("", self.repositories.workspace_status().strip())

    def test_DropsADeletedFile(self):
        workspace_file = os.path.join(self.repositories.workspace, "Modules", "b.txt")
        self.repositories.svn("delete", workspace_file)
        self.repositories.svn("changelist", "CL", workspace_file)
        self.repositories.deliver()
        self.repositories.commit_main()
        self.repositories.landed()
        self.assertFalse(os.path.exists(workspace_file))
        self.assertEqual("", self.repositories.workspace_status().strip())

    def test_DropsAPropertyOnlyChange(self):
        workspace_file = os.path.join(self.repositories.workspace, "Modules", "a.txt")
        self.repositories.svn("propset", "svn:eol-style", "native", workspace_file)
        self.repositories.svn("changelist", "CL", workspace_file)
        self.repositories.deliver()
        self.repositories.commit_main()
        self.repositories.landed()
        self.assertEqual("native", self.repositories.svn("propget", "svn:eol-style", workspace_file).strip())
        self.assertEqual("", self.repositories.workspace_status().strip())

    def test_UsesTheLatestOfTwoDeliveries(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "first\n")
        self.repositories.deliver()
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "second\n")
        self.repositories.deliver()
        self.repositories.commit_main()
        self.repositories.landed()
        self.assertEqual(numbered_lines(29) + "second\n",
                         self.repositories.read(self.repositories.workspace, "Modules/a.txt"))

    def test_ForgetsTheChangelistsRecord(self):
        self.edit_deliver_and_commit()
        self.repositories.landed()
        self.assertFalse(os.path.exists(self.repositories.record_path))

    def test_KeepsTheRecordsOfOtherChangelists(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "b.txt"), numbered_lines(29) + "two\n", "Other")
        self.repositories.deliver("Other")
        self.repositories.svn("revert", os.path.join(self.repositories.main, "Modules", "b.txt"))
        self.edit_deliver_and_commit()
        self.repositories.landed()
        self.assertIsNone(self.repositories.record_for("CL"))
        self.assertIsNotNone(self.repositories.record_for("Other"))

    def test_LeavesAnUndeliveredChangelistAlone(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "b.txt"), numbered_lines(29) + "two\n", "Other")
        self.edit_deliver_and_commit()
        self.repositories.landed()
        self.assertEqual(numbered_lines(29) + "two\n",
                         self.repositories.read(self.repositories.workspace, "Modules/b.txt"))
        status = self.repositories.svn("status", os.path.join(self.repositories.workspace, "Modules"), "--cl", "Other")
        self.assertIn("b.txt", status)

    def test_ADryRunChangesNothing(self):
        self.edit_deliver_and_commit()
        before = self.repositories.workspace_status()
        files, messages = self.repositories.landed(dry_run=True)
        self.assertEqual(1, len(files))
        self.assertEqual(before, self.repositories.workspace_status())
        self.assertIsNotNone(self.repositories.record_for())
        self.assertTrue(any("committed in main" in message for message in messages))


class MergedUpdateLandingTests(DeliverFixture):
    """A workspace updated between delivering and landing has the commits of others merged into its copies."""

    def deliver_edit_while_others_commit_elsewhere_in_the_file(self):
        edited = numbered_lines(29) + "mine\n"
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), edited)
        self.repositories.deliver()
        upstream = numbered_lines(30).replace("line 1\n", "changed upstream\n")
        self.repositories.write(self.repositories.seed_modules, "a.txt", upstream)
        self.repositories.svn("commit", "-m", "upstream", cwd=self.repositories.seed_modules)
        self.repositories.svn("update", os.path.join(self.repositories.workspace, "Modules"))
        self.repositories.svn("update", os.path.join(self.repositories.main, "Modules"))
        return upstream.replace("line 30\n", "") + "line 29\nmine\n"

    def test_LandsAFileWhoseWorkspaceCopyHasAnUpstreamEditMergedIn(self):
        self.deliver_edit_while_others_commit_elsewhere_in_the_file()
        self.repositories.commit_main()
        files, messages = self.repositories.landed()
        self.assertEqual(1, len(files))
        workspace_text = self.repositories.read(self.repositories.workspace, "Modules/a.txt")
        self.assertEqual(self.repositories.read(self.repositories.main, "Modules/a.txt"), workspace_text)
        self.assertIn("changed upstream", workspace_text)
        self.assertIn("mine", workspace_text)
        self.assertEqual("", self.repositories.workspace_status().strip())

    def test_StillRefusesAMergedFileWhoseContentMainDoesNotHave(self):
        self.deliver_edit_while_others_commit_elsewhere_in_the_file()
        self.repositories.commit_main()
        self.repositories.write(self.repositories.workspace, os.path.join("Modules", "a.txt"), "later edit\n")
        before = self.repositories.workspace_status()
        with self.assertRaises(deliver.DeliveryRefused) as caught:
            self.repositories.landed()

        self.assertIn("edited after it was delivered", str(caught.exception))
        self.assertEqual(before, self.repositories.workspace_status())
        self.assertEqual("later edit\n", self.repositories.read(self.repositories.workspace, "Modules/a.txt"))

    def test_LandsAFileWhoseWorkspacePropertiesAreTheOnesMainCommitted(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "mine\n")
        self.repositories.deliver()
        workspace_file = os.path.join(self.repositories.workspace, "Modules", "a.txt")
        main_file = os.path.join(self.repositories.main, "Modules", "a.txt")
        self.repositories.svn("propset", "svn:mime-type", "text/plain", workspace_file)
        self.repositories.svn("propset", "svn:mime-type", "text/plain", main_file)
        self.repositories.commit_main()
        files, messages = self.repositories.landed()
        self.assertEqual(1, len(files))
        self.assertEqual("text/plain", self.repositories.svn("propget", "svn:mime-type", workspace_file).strip())
        self.assertEqual("", self.repositories.workspace_status().strip())


class AbsorbedChangelistLandingTests(DeliverFixture):
    """Landing one changelist updates the workspace past others that were committed with it."""

    def deliver_two_changelists(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "one\n", "CL1")
        self.repositories.deliver("CL1")
        self.repositories.edit_in_workspace(os.path.join("Modules", "b.txt"), numbered_lines(29) + "two\n", "CL2")
        self.repositories.add_to_workspace(os.path.join("Modules", "n.txt"), "new\n", "CL2")
        self.repositories.deliver("CL2")

    def land_the_first_of_two_committed_together(self):
        self.deliver_two_changelists()
        self.repositories.commit_main()
        self.repositories.landed("CL1")

    def test_LandsAChangelistTheFirstLandingUpdatedPast(self):
        self.land_the_first_of_two_committed_together()
        files, messages = self.repositories.landed("CL2")
        self.assertEqual(2, len(files))
        self.assertIsNone(self.repositories.record_for("CL2"))
        self.assertEqual("", self.repositories.workspace_status().strip())
        self.assertEqual("", self.repositories.status_of_main().strip())

    def test_LeavesTheCommittedContentInTheWorkspace(self):
        self.land_the_first_of_two_committed_together()
        self.repositories.landed("CL2")
        self.assertEqual(numbered_lines(29) + "two\n", self.repositories.read(self.repositories.workspace, "Modules/b.txt"))
        self.assertEqual("new\n", self.repositories.read(self.repositories.workspace, "Modules/n.txt"))

    def test_ADryRunReportsItAndChangesNothing(self):
        self.land_the_first_of_two_committed_together()
        before = self.repositories.workspace_status()
        files, messages = self.repositories.landed("CL2", dry_run=True)
        self.assertEqual(2, len(files))
        self.assertEqual(before, self.repositories.workspace_status())
        self.assertIsNotNone(self.repositories.record_for("CL2"))

    def test_RefusesWhenMainHasNotCommittedIt(self):
        self.deliver_two_changelists()
        self.repositories.svn("commit", "-m", "only the first", os.path.join(self.repositories.main, "Modules", "a.txt"))
        self.repositories.landed("CL1")
        before = self.repositories.workspace_status()
        files_pending = self.repositories.svn("status", os.path.join(self.repositories.main, "Modules"))
        self.assertIn("b.txt", files_pending)
        with self.assertRaises(deliver.DeliveryRefused) as caught:
            self.repositories.landed("CL2")

        self.assertIn("hasn't been committed", str(caught.exception))
        self.assertEqual(before, self.repositories.workspace_status())
        self.assertIsNotNone(self.repositories.record_for("CL2"))

    def test_LandsAChangelistWhoseWorkspaceCopyWasRevertedAfterMainCommittedIt(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "one\n")
        self.repositories.deliver()
        self.repositories.commit_main()
        self.repositories.svn("revert", os.path.join(self.repositories.workspace, "Modules", "a.txt"))
        files, messages = self.repositories.landed()
        self.assertEqual(1, len(files))
        self.assertEqual(numbered_lines(29) + "one\n", self.repositories.read(self.repositories.workspace, "Modules/a.txt"))
        self.assertEqual("", self.repositories.workspace_status().strip())

    def test_AFileWithOtherPendingChangesInTheWorkspaceIsNotAbsorbed(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "one\n")
        self.repositories.deliver()
        self.repositories.commit_main()
        self.repositories.svn("changelist", "--remove", os.path.join(self.repositories.workspace, "Modules", "a.txt"))
        before = self.repositories.workspace_status()
        with self.assertRaises(deliver.DeliveryRefused) as caught:
            self.repositories.landed()

        self.assertIn("has left the changelist", str(caught.exception))
        self.assertEqual(before, self.repositories.workspace_status())


class LandedRefusalTests(DeliverFixture):
    def deliver_edit(self, text="edited\n"):
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + text)
        self.repositories.deliver()

    def assert_refused(self, expected_text, name="CL"):
        """Checks landed() refuses with a message holding `expected_text`, and that the workspace stays as it was."""
        before = self.repositories.workspace_status()
        with self.assertRaises(deliver.DeliveryRefused) as caught:
            self.repositories.landed(name)

        self.assertIn(expected_text, str(caught.exception))
        self.assertEqual(before, self.repositories.workspace_status())

    def test_RefusesAChangelistThatWasNeverDelivered(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "a.txt"), numbered_lines(29) + "edited\n")
        self.assert_refused("No delivery of 'CL' is recorded")

    def test_RefusesAChangelistWhoseWorkspaceCopyWasRevertedBeforeMainCommittedIt(self):
        self.deliver_edit()
        self.repositories.svn("revert", os.path.join(self.repositories.workspace, "Modules", "a.txt"))
        self.assert_refused("hasn't been committed")
        self.assertIsNotNone(self.repositories.record_for())

    def test_RefusesAnEditNotYetCommitted(self):
        self.deliver_edit()
        self.assert_refused("hasn't been committed")
        self.assertIsNotNone(self.repositories.record_for())

    def test_RefusesAnAddedFileNotYetCommitted(self):
        workspace_modules = os.path.join(self.repositories.workspace, "Modules")
        self.repositories.write(workspace_modules, "n.txt", "new\n")
        self.repositories.svn("add", os.path.join(workspace_modules, "n.txt"))
        self.repositories.svn("changelist", "CL", os.path.join(workspace_modules, "n.txt"))
        self.repositories.deliver()
        self.assert_refused("hasn't been committed")

    def test_RefusesADeletionNotYetCommitted(self):
        workspace_file = os.path.join(self.repositories.workspace, "Modules", "b.txt")
        self.repositories.svn("delete", workspace_file)
        self.repositories.svn("changelist", "CL", workspace_file)
        self.repositories.deliver()
        self.assert_refused("deletion hasn't been committed")

    def test_RefusesADeliveryThatWasRevertedInMain(self):
        self.deliver_edit()
        self.repositories.svn("revert", os.path.join(self.repositories.main, "Modules", "a.txt"))
        self.assert_refused("no commit in main since it was delivered")

    def test_RefusesAnAddedFileMissingFromMain(self):
        workspace_modules = os.path.join(self.repositories.workspace, "Modules")
        self.repositories.write(workspace_modules, "n.txt", "new\n")
        self.repositories.svn("add", os.path.join(workspace_modules, "n.txt"))
        self.repositories.svn("changelist", "CL", os.path.join(workspace_modules, "n.txt"))
        self.repositories.deliver()
        main_file = os.path.join(self.repositories.main, "Modules", "n.txt")
        self.repositories.svn("revert", main_file)
        os.remove(main_file)
        self.assert_refused("isn't in main")

    def test_RefusesAFileEditedAfterTheDelivery(self):
        self.deliver_edit()
        self.repositories.commit_main()
        self.repositories.write(self.repositories.workspace, os.path.join("Modules", "a.txt"), "edited again\n")
        self.assert_refused("was edited after it was delivered")
        self.assertEqual("edited again\n", self.repositories.read(self.repositories.workspace, "Modules/a.txt"))

    def test_RefusesAFileWhosePropertiesChangedAfterTheDelivery(self):
        workspace_file = os.path.join(self.repositories.workspace, "Modules", "a.txt")
        self.repositories.svn("propset", "note", "one", workspace_file)
        self.repositories.svn("changelist", "CL", workspace_file)
        self.repositories.deliver()
        self.repositories.commit_main()
        self.repositories.svn("propset", "note", "two", workspace_file)
        self.assert_refused("had its properties changed after it was delivered")

    def test_RefusesAFileWhoseStatusChangedAfterTheDelivery(self):
        self.deliver_edit()
        self.repositories.commit_main()
        self.repositories.svn("propset", "svn:eol-style", "native",
                              os.path.join(self.repositories.workspace, "Modules", "a.txt"))
        self.assert_refused("was 'M ' when delivered and is 'MM' now")

    def test_RefusesAFileAddedToTheChangelistAfterTheDelivery(self):
        self.deliver_edit()
        self.repositories.commit_main()
        self.repositories.edit_in_workspace(os.path.join("Modules", "b.txt"), numbered_lines(29) + "later\n")
        self.assert_refused("joined the changelist after it was delivered")

    def test_RefusesAFileTakenOutOfTheChangelistAfterTheDelivery(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "b.txt"), numbered_lines(29) + "two\n")
        self.deliver_edit()
        self.repositories.commit_main()
        self.repositories.svn("changelist", "--remove", os.path.join(self.repositories.workspace, "Modules", "b.txt"))
        self.assert_refused("has left the changelist")

    def test_ListsEveryProblemAtOnce(self):
        self.repositories.edit_in_workspace(os.path.join("Modules", "b.txt"), numbered_lines(29) + "two\n")
        self.deliver_edit()
        with self.assertRaises(deliver.DeliveryRefused) as caught:
            self.repositories.landed()

        self.assertIn("a.txt", str(caught.exception))
        self.assertIn("b.txt", str(caught.exception))


class ChunksTests(unittest.TestCase):
    def test_SplitsAListIntoPiecesOfAtMostTheSize(self):
        self.assertEqual([[1, 2], [3, 4], [5]], list(deliver.chunks([1, 2, 3, 4, 5], size=2)))

    def test_AListOfExactlyTheSizeIsOnePiece(self):
        self.assertEqual([[1, 2]], list(deliver.chunks([1, 2], size=2)))

    def test_AnEmptyListGivesNoPieces(self):
        self.assertEqual([], list(deliver.chunks([], size=2)))


class ParseStatusTests(unittest.TestCase):
    def test_AssignsEachItemTheChangelistItWasListedUnder(self):
        output = "M       a.txt\n\n--- Changelist 'One':\nM       b.txt\nA  +    c.txt\n"
        self.assertEqual([("M      ", os.path.normpath("a.txt"), None), ("M      ", os.path.normpath("b.txt"), "One"),
                          ("A  +   ", os.path.normpath("c.txt"), "One")],
                         [tuple(entry) for entry in deliver.parse_status(output)])

    def test_IgnoresLinesThatAreNotItems(self):
        self.assertEqual([], deliver.parse_status("Performing status on external item at 'x':\n\n"))

    def test_IgnoresTheLinesThatDescribeAMove(self):
        output = ("D       a.txt\n        > moved to b.txt\nA  +    b.txt\n        > moved from a.txt\n"
                  "      >   local file edit, incoming file delete\n")
        self.assertEqual([os.path.normpath("a.txt"), os.path.normpath("b.txt")],
                         [entry.path for entry in deliver.parse_status(output)])


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