<commit_before>"""
Tests for the Voluminous CLI.
"""

from string import letters

from hypothesis import given
from hypothesis.strategies import sets, text
from twisted.trial.unittest import TestCase
from twisted.python.filepath import FilePath
from dvol import VoluminousOptions, VolumeAlreadyExists, Voluminous
from twisted.python.usage import UsageError

class EmptyContainers(object):
    def get_related_containers(self, volume):
        return []

class NullLock(object):
    containers = EmptyContainers()
    def acquire(self, volume):
        return
    def release(self, volume):
        return

class VoluminousTests(TestCase):
    def setUp(self):
        self.tmpdir = FilePath(self.mktemp())
        self.tmpdir.makedirs()
        self.patch(Voluminous, "lockFactory", NullLock)

    def test_create_volume(self):
        # TODO test volume names with '/' in them - they should not end up
        # making nested heirarchy
        dvol = VoluminousOptions()
        dvol.parseOptions(["-p", self.tmpdir.path, "init", "foo"])
        self.assertTrue(self.tmpdir.child("foo").exists())
        self.assertTrue(self.tmpdir.child("foo").child("branches")
                .child("master").exists())
        self.assertEqual(dvol.voluminous.getOutput(),
                ["Created volume foo", "Created branch foo/master"])

    def test_create_volume_already_exists(self):
        dvol = VoluminousOptions()
        dvol.parseOptions(["-p", self.tmpdir.path, "init", "foo"])
        self.assertRaises(VolumeAlreadyExists,
                dvol.parseOptions, ["-p", self.tmpdir.path, "init", "foo"])
        self.assertEqual(dvol.voluminous.getOutput(),
                ["Error: volume foo already exists"])

    def test_commit_no_message_raises_error(self):
        dvol = VoluminousOptions()
        dvol.parseOptions(["-p", self.tmpdir.path, "init", "foo"])
        self.assertRaises(UsageError,
                dvol.parseOptions, ["-p", self.tmpdir.path, "commit"])

    def test_commit_volume(self):
        # TODO need to assert that containers using this volume get stopped
        # and started around commits
        # TODO test snapshotting nonexistent volume
        dvol = VoluminousOptions()
        dvol.parseOptions(["-p", self.tmpdir.path, "init", "foo"])
        volume = self.tmpdir.child("foo")
        volume.child("branches").child("master").child(
            "file.txt").setContent("hello!")
        dvol.parseOptions(["-p", self.tmpdir.path,
            "commit", "-m", "hello from 30,000 ft"])
        commitId = dvol.voluminous.getOutput()[-1]
        commit = volume.child("commits").child(commitId)
        self.assertTrue(commit.exists())
        self.assertTrue(commit.child("file.txt").exists())
        self.assertEqual(commit.child("file.txt").getContent(), "hello!")

    def test_list_empty_volumes(self):
        dvol = VoluminousOptions()
        dvol.parseOptions(["-p", self.tmpdir.path, "list"])
        self.assertEqual(dvol.voluminous.getOutput(), ["  VOLUME   BRANCH   CONTAINERS "])

    # XXX: Fix the bug about empty volume names
    # XXX: Handle unicode / weird volume names by rejecting them in dvol
    # XXX Impose a maximum volume name length (at least so rendering is easy!)
    @given(volume_names=sets(text(alphabet=letters, min_size=1, max_size=112), min_size=1, average_size=10).map(list))
    def test_list_multi_volumes(self, volume_names):
        # XXX: This initializes duplicate volumes. I wonder what happens then.
        tmpdir = FilePath(self.mktemp())
        tmpdir.makedirs()

        dvol = VoluminousOptions()
        for volume_name in volume_names:
            dvol.parseOptions(["-p", tmpdir.path, "init", volume_name])
        dvol.parseOptions(["-p", tmpdir.path, "list"])

        lines = dvol.voluminous.getOutput()[0].split("\n")
        header, rest = lines[0], lines[1:]
        expected_volumes = [[volume_name, 'master'] for volume_name in volume_names]
        # `init` activates the volume, so the last initialized volume is the
        # active one.
        expected_volumes[-1] = ['*', expected_volumes[-1][0], expected_volumes[-1][1]]
        self.assertEqual(['VOLUME', 'BRANCH', 'CONTAINERS'], header.split())
        self.assertEqual(
            sorted(expected_volumes),
            sorted([line.split() for line in rest]),
        )

    def test_log(self):
        dvol = VoluminousOptions()
        dvol.parseOptions(["-p", self.tmpdir.path,
            "init", "foo"])
        dvol.parseOptions(["-p", self.tmpdir.path,
            "commit", "-m", "oi"])
        dvol.parseOptions(["-p", self.tmpdir.path,
            "commit", "-m", "you"])
        dvol.parseOptions(["-p", self.tmpdir.path,
            "log"])
        actual = dvol.voluminous.getOutput()[-1]
        expected = (
            "commit\n"
            "Author:\n"
            "Date:\n"
            "\n"
            "    you\n"<commit_msg>jml's changes<commit_after>"""
Tests for the Voluminous CLI.
"""

from string import letters

from hypothesis import given
from hypothesis.strategies import binary, characters, sets, text
from twisted.trial.unittest import TestCase
from twisted.python.filepath import FilePath
from dvol import VoluminousOptions, VolumeAlreadyExists, Voluminous
from twisted.python.usage import UsageError

class EmptyContainers(object):
    def get_related_containers(self, volume):
        return []

class NullLock(object):
    containers = EmptyContainers()
    def acquire(self, volume):
        return
    def release(self, volume):
        return


def path_segments():
    """
    Strategy for generating path segments that we support.
    """
    # XXX: Fix the bug about empty volume names
    # XXX: Handle unicode / weird volume names by rejecting them in dvol
    # XXX: Impose a maximum volume name length (at least so rendering is easy!)
    # XXX: How do we handle case-insensitive file systems?
    return text(alphabet=letters, min_size=1, max_size=40).map(lambda t: t.lower())


volume_names = path_segments
branch_names = path_segments


class VoluminousTests(TestCase):
    def setUp(self):
        self.tmpdir = FilePath(self.mktemp())
        self.tmpdir.makedirs()
        self.patch(Voluminous, "lockFactory", NullLock)

    def test_create_volume(self):
        # TODO test volume names with '/' in them - they should not end up
        # making nested heirarchy
        dvol = VoluminousOptions()
        dvol.parseOptions(["-p", self.tmpdir.path, "init", "foo"])
        self.assertTrue(self.tmpdir.child("foo").exists())
        self.assertTrue(self.tmpdir.child("foo").child("branches")
                .child("master").exists())
        self.assertEqual(dvol.voluminous.getOutput(),
                ["Created volume foo", "Created branch foo/master"])

    def test_create_volume_already_exists(self):
        dvol = VoluminousOptions()
        dvol.parseOptions(["-p", self.tmpdir.path, "init", "foo"])
        self.assertRaises(VolumeAlreadyExists,
                dvol.parseOptions, ["-p", self.tmpdir.path, "init", "foo"])
        self.assertEqual(dvol.voluminous.getOutput(),
                ["Error: volume foo already exists"])

    def test_commit_no_message_raises_error(self):
        dvol = VoluminousOptions()
        dvol.parseOptions(["-p", self.tmpdir.path, "init", "foo"])
        self.assertRaises(UsageError,
                dvol.parseOptions, ["-p", self.tmpdir.path, "commit"])

    def test_commit_volume(self):
        # TODO need to assert that containers using this volume get stopped
        # and started around commits
        # TODO test snapshotting nonexistent volume
        dvol = VoluminousOptions()
        dvol.parseOptions(["-p", self.tmpdir.path, "init", "foo"])
        volume = self.tmpdir.child("foo")
        volume.child("branches").child("master").child(
            "file.txt").setContent("hello!")
        dvol.parseOptions(["-p", self.tmpdir.path,
            "commit", "-m", "hello from 30,000 ft"])
        commitId = dvol.voluminous.getOutput()[-1]
        commit = volume.child("commits").child(commitId)
        self.assertTrue(commit.exists())
        self.assertTrue(commit.child("file.txt").exists())
        self.assertEqual(commit.child("file.txt").getContent(), "hello!")

    def test_list_empty_volumes(self):
        dvol = VoluminousOptions()
        dvol.parseOptions(["-p", self.tmpdir.path, "list"])
        self.assertEqual(dvol.voluminous.getOutput(), ["  VOLUME   BRANCH   CONTAINERS "])

    @given(volumes=sets(volume_names(), min_size=1, average_size=10).map(list))
    def test_list_multi_volumes(self, volumes):
        tmpdir = FilePath(self.mktemp())
        tmpdir.makedirs()

        dvol = VoluminousOptions()
        for name in volumes:
            dvol.parseOptions(["-p", tmpdir.path, "init", name])
        dvol.parseOptions(["-p", tmpdir.path, "list"])

        lines = dvol.voluminous.getOutput()[0].split("\n")
        header, rest = lines[0], lines[1:]
        expected_volumes = [[name, 'master'] for name in volumes]
        # `init` activates the volume, so the last initialized volume is the
        # active one.
        expected_volumes[-1] = ['*', expected_volumes[-1][0], expected_volumes[-1][1]]
        self.assertEqual(['VOLUME', 'BRANCH', 'CONTAINERS'], header.split())
        self.assertEqual(
            sorted(expected_volumes),
            sorted([line.split() for line in rest]),
        )

    @given(volume_name=volume_names(), branch_name=branch_names(),
           commit_message=text(characters(max_codepoint=127), min_size=1),
           filename=path_segments(), content=binary())
    def test_non_standard_branch(self, volume_name, branch_name, commit_message, filename,
                                 content):
        tmpdir = FilePath(self.mktemp())
        tmpdir.makedirs()

        dvol = VoluminousOptions()
        dvol.parseOptions(['-p', tmpdir.path, 'init', volume_name])
        volume = tmpdir.child(volume_name)
        volume.child("branches").child("master").child(filename).setContent(content)
        dvol.parseOptions(["-p", tmpdir.path, "commit", "-m", commit_message])
        dvol.parseOptions(["-p", tmpdir.path, "checkout", "-b", branch_name])
        dvol.parseOptions(["-p", tmpdir.path, "list"])
        lines = dvol.voluminous.getOutput()[0].split("\n")
        header, rest = lines[0], lines[1:]
        self.assertEqual(['VOLUME', 'BRANCH', 'CONTAINERS'], header.split())
        self.assertEqual(
            [['*', volume_name, branch_name]], [line.split() for line in rest])

    def test_log(self):
        dvol = VoluminousOptions()
        dvol.parseOptions(["-p", self.tmpdir.path,
            "init", "foo"])
        dvol.parseOptions(["-p", self.tmpdir.path,
            "commit", "-m", "oi"])
        dvol.parseOptions(["-p", self.tmpdir.path,
            "commit", "-m", "you"])
        dvol.parseOptions(["-p", self.tmpdir.path,
            "log"])
        actual = dvol.voluminous.getOutput()[-1]
        expected = (
            "commit\n"
            "Author:\n"
            "Date:\n"
            "\n"
            "    you\n"
{'fraction_non_alphanumeric': 0.13998919697515305, 'fraction_numerical': 0.004501260352898812, 'mean_word_length': 2.797948717948718, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6768169', 'n_tokens_mistral': 3386, 'n_tokens_neox': 3237, 'n_words': 724}