# -*- Mode:Python; indent-tabs-mode:nil; tab-width:4; encoding:utf-8 -*-
#
# Copyright 2002 Ben Escoto
# Copyright 2007 Kenneth Loafman
#
# This file is part of duplicity.
#
# Duplicity is free software; you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the
# Free Software Foundation; either version 2 of the License, or (at your
# option) any later version.
#
# Duplicity is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with duplicity; if not, write to the Free Software Foundation,
# Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA


import glob
import os
import platform
import subprocess
import unittest

import pytest

from testing.functional import _runtest_dir
from testing.functional import FunctionalTestCase


class RestartTest(FunctionalTestCase):
    """
    Test checkpoint/restart using duplicity binary
    """

    def test_basic_checkpoint_restart(self):
        """
        Test basic Checkpoint/Restart
        """
        self.make_largefiles()
        self.backup("full", f"{_runtest_dir}/testfiles/largefiles", fail=1)
        self.backup("full", f"{_runtest_dir}/testfiles/largefiles")
        self.verify(f"{_runtest_dir}/testfiles/largefiles")

    @pytest.mark.slow
    def test_multiple_checkpoint_restart(self):
        """
        Test multiple Checkpoint/Restart
        """
        self.make_largefiles()
        self.backup("full", f"{_runtest_dir}/testfiles/largefiles", fail=1)
        self.backup("full", f"{_runtest_dir}/testfiles/largefiles", fail=2)
        self.backup("full", f"{_runtest_dir}/testfiles/largefiles", fail=3)
        self.backup("full", f"{_runtest_dir}/testfiles/largefiles")
        self.verify(f"{_runtest_dir}/testfiles/largefiles")

    def test_first_volume_failure(self):
        """
        Test restart when no volumes are available on the remote.
        Caused when duplicity fails before the first transfer.
        """
        self.make_largefiles()
        self.backup("full", f"{_runtest_dir}/testfiles/largefiles", fail=1)
        os.system(f"rm {_runtest_dir}/testfiles/output/duplicity-full*difftar*")
        self.backup("full", f"{_runtest_dir}/testfiles/largefiles")
        self.verify(f"{_runtest_dir}/testfiles/largefiles")

    def test_multi_volume_failure(self):
        """
        Test restart when fewer volumes are available on the remote
        than the local manifest has on record.  Caused when duplicity
        fails the last queued transfer(s).
        """
        self.make_largefiles()
        self.backup("full", f"{_runtest_dir}/testfiles/largefiles", fail=3)
        os.system(f"rm {_runtest_dir}/testfiles/output/duplicity-full*vol[23].difftar*")
        self.backup("full", f"{_runtest_dir}/testfiles/largefiles")
        self.verify(f"{_runtest_dir}/testfiles/largefiles")

    def test_restart_encrypt_without_password(self):
        """
        Test that we can successfully restart a encrypt-key-only backup without
        providing a password for it. (Normally, we'd need to decrypt the first
        volume, but there is special code to skip that with an encrypt key.)
        """
        self.set_environ("PASSPHRASE", None)
        self.set_environ("SIGN_PASSPHRASE", None)
        self.make_largefiles()
        enc_opts = ["--encrypt-key", self.encrypt_key1]
        self.backup("full", f"{_runtest_dir}/testfiles/largefiles", options=enc_opts, fail=2)
        self.backup("full", f"{_runtest_dir}/testfiles/largefiles", options=enc_opts)

        self.set_environ("PASSPHRASE", self.sign_passphrase)
        self.verify(f"{_runtest_dir}/testfiles/largefiles")

    def test_restart_sign_and_encrypt(self):
        """
        Test restarting a backup using same key for sign and encrypt
        https://bugs.launchpad.net/duplicity/+bug/946988
        """
        self.make_largefiles()
        enc_opts = ["--sign-key", self.sign_key, "--encrypt-key", self.sign_key]
        self.backup("full", f"{_runtest_dir}/testfiles/largefiles", options=enc_opts, fail=2)
        self.backup("full", f"{_runtest_dir}/testfiles/largefiles", options=enc_opts)
        self.verify(f"{_runtest_dir}/testfiles/largefiles")

    def test_restart_sign_and_hidden_encrypt(self):
        """
        Test restarting a backup using same key for sign and encrypt (hidden key id)
        https://bugs.launchpad.net/duplicity/+bug/946988
        """
        self.make_largefiles()
        enc_opts = ["--sign-key", self.sign_key, "--hidden-encrypt-key", self.sign_key]
        self.backup("full", f"{_runtest_dir}/testfiles/largefiles", options=enc_opts, fail=2)
        self.backup("full", f"{_runtest_dir}/testfiles/largefiles", options=enc_opts)
        self.verify(f"{_runtest_dir}/testfiles/largefiles")

    def test_last_file_missing_in_middle(self):
        """
        Test restart when the last file being backed up is missing on restart.
        Caused when the user deletes a file after a failure.  This test puts
        the file in the middle of the backup, with files following.
        """
        self.make_largefiles()
        self.backup("full", f"{_runtest_dir}/testfiles/largefiles", fail=3)
        assert not os.system(f"rm {_runtest_dir}/testfiles/largefiles/file2")
        self.backup("full", f"{_runtest_dir}/testfiles/largefiles")
        self.verify(f"{_runtest_dir}/testfiles/largefiles")

    @unittest.skip("Flaky test because it relies on knowing how many volumes the source files will be split into")
    @unittest.skipIf(
        platform.machine() in ["ppc64el", "ppc64le"],
        "See https://gitlab.com/duplicity/duplicity/-/issues/820",
    )
    def test_last_file_missing_at_end(self):
        """
        Test restart when the last file being backed up is missing on restart.
        Caused when the user deletes a file after a failure.  This test puts
        the file at the end of the backup, with no files following.
        """
        self.make_largefiles()
        self.backup("full", f"{_runtest_dir}/testfiles/largefiles", fail=6)
        assert not os.system(f"rm {_runtest_dir}/testfiles/largefiles/file3")
        self.backup("full", f"{_runtest_dir}/testfiles/largefiles")
        self.verify(f"{_runtest_dir}/testfiles/largefiles")

    @pytest.mark.slow
    def test_restart_incremental(self):
        """
        Test restarting an incremental backup
        """
        self.make_largefiles()
        self.backup(
            "full",
            f"{_runtest_dir}/testfiles/dir1",
            options=["--allow-source-mismatch"],
        )
        self.backup(
            "inc",
            f"{_runtest_dir}/testfiles/largefiles",
            fail=2,
            options=["--allow-source-mismatch"],
        )
        self.backup(
            "inc",
            f"{_runtest_dir}/testfiles/largefiles",
            options=["--allow-source-mismatch"],
        )
        self.verify(f"{_runtest_dir}/testfiles/largefiles")

    def make_fake_second_volume(self, name):
        """
        Takes a successful backup and pretend that we interrupted a backup
        after two-volumes.  (This is because we want to be able to model
        restarting the second volume and duplicity deletes the last volume
        found because it may have not finished uploading.)
        """
        # First, confirm that we have signs of a successful backup
        self.assertEqual(len(glob.glob(f"{_runtest_dir}/testfiles/output/*.manifest*")), 1)
        self.assertEqual(len(glob.glob(f"{_runtest_dir}/testfiles/output/*.sigtar*")), 1)
        self.assertEqual(len(glob.glob(f"{_runtest_dir}/testfiles/cache/{name}/*")), 2)
        self.assertEqual(len(glob.glob(f"{_runtest_dir}/testfiles/cache/{name}/*.manifest*")), 1)
        self.assertEqual(len(glob.glob(f"{_runtest_dir}/testfiles/cache/{name}/*.sigtar*")), 1)
        # Alright, everything is in order; fake a second interrupted volume
        assert not os.system(f"rm {_runtest_dir}/testfiles/output/*.manifest*")
        assert not os.system(f"rm {_runtest_dir}/testfiles/output/*.sigtar*")
        assert not os.system(f"rm -f {_runtest_dir}/testfiles/output/*.vol[23456789].*")
        assert not os.system(f"rm -f {_runtest_dir}/testfiles/output/*.vol1[^.]+.*")
        self.assertEqual(len(glob.glob(f"{_runtest_dir}/testfiles/output/*.difftar*")), 1)
        assert not os.system(f"rm {_runtest_dir}/testfiles/cache/{name}/*.sigtar*")
        assert not os.system(
            f"cp {_runtest_dir}/testfiles/output/*.difftar* "
            + f"`ls {_runtest_dir}/testfiles/output/*.difftar* | "
            + " sed 's|vol1|vol2|'`"
        )
        assert not os.system(
            f"head -n6 {_runtest_dir}/testfiles/cache/{name}/*.manifest > "
            + f"{_runtest_dir}/testfiles/cache/{name}/"
            + f"`basename {_runtest_dir}/testfiles/cache/{name}/*.manifest`"
            + ".part"
        )
        assert not os.system(f"rm {_runtest_dir}/testfiles/cache/{name}/*.manifest")
        assert not os.system(
            f"""echo 'Volume 2:
    StartingPath   foo
    EndingPath     bar
    Hash SHA1 sha1' >> {_runtest_dir}/testfiles/cache/{name}/*.manifest.part"""
        )

    def test_split_after_small(self):
        """
        If we restart right after a volume that ended with a small
        (one-block) file, make sure we restart in the right place.
        """
        source = f"{_runtest_dir}/testfiles/largefiles"
        assert not os.system(f"mkdir -p {source}")
        assert not os.system(f"echo hello > {source}/file1")
        self.backup("full", source, options=["--name=backup1"])
        # Fake an interruption
        self.make_fake_second_volume("backup1")
        # Add new file
        assert not os.system(f"cp {source}/file1 {source}/newfile")
        # 'restart' the backup
        self.backup("full", source, options=["--name=backup1"])
        # Confirm we actually resumed the previous backup
        self.assertEqual(len(os.listdir(f"{_runtest_dir}/testfiles/output")), 4)
        # Now make sure everything is byte-for-byte the same once restored
        self.restore()
        assert not os.system(f"diff -r {source} {_runtest_dir}/testfiles/restore_out")

    def test_split_after_large(self):
        """
        If we restart right after a volume that ended with a large
        (multi-block) file, make sure we restart in the right place.
        """
        source = f"{_runtest_dir}/testfiles/largefiles"
        self.make_largefiles(count=1, size=1)
        self.backup("full", source, options=["--volsize=5", "--name=backup1"])
        # Fake an interruption
        self.make_fake_second_volume("backup1")
        # Add new file
        assert not os.system(f"cp {source}/file1 {source}/newfile")
        # 'restart' the backup
        self.backup("full", source, options=["--volsize=5", "--name=backup1"])
        # Confirm we actually resumed the previous backup
        self.assertEqual(len(os.listdir(f"{_runtest_dir}/testfiles/output")), 4)
        # Now make sure everything is byte-for-byte the same once restored
        self.restore()
        assert not os.system(f"diff -r %s {_runtest_dir}/testfiles/restore_out" % source)

    def test_split_inside_large(self):
        """
        If we restart right after a volume that ended inside of a large
        (multi-block) file, make sure we restart in the right place.
        """
        source = f"{_runtest_dir}/testfiles/largefiles"
        self.make_largefiles(count=1, size=3)
        self.backup("full", source, options=["--name=backup1"])
        # Fake an interruption
        self.make_fake_second_volume("backup1")
        # 'restart' the backup
        self.backup("full", source, options=["--name=backup1"])
        # Now make sure everything is byte-for-byte the same once restored
        self.restore()
        assert not os.system(f"diff -r {source} {_runtest_dir}/testfiles/restore_out")

    def test_new_file(self):
        """
        If we restart right after a volume, but there are new files that would
        have been backed up earlier in the volume, make sure we don't wig out.
        (Expected result is to ignore new, ealier files, but pick up later
        ones.)
        """
        source = f"{_runtest_dir}/testfiles/largefiles"
        self.make_largefiles(count=1, size=1)
        self.backup("full", source, options=["--name=backup1"])
        # Fake an interruption
        self.make_fake_second_volume("backup1")
        # Add new files, earlier and later in filename sort order
        assert not os.system(f"echo hello > {source}/a")
        assert not os.system(f"echo hello > {source}/z")
        # 'restart' the backup
        self.backup("full", source, options=["--name=backup1"])
        # Now make sure everything is the same once restored, except 'a'
        self.restore()
        assert not os.system(f"test ! -e {_runtest_dir}/testfiles/restore_out/a")
        assert not os.system(f"diff {source}/file1 {_runtest_dir}/testfiles/restore_out/file1")
        assert not os.system(f"diff {source}/z {_runtest_dir}/testfiles/restore_out/z")

    def test_changed_source_dangling_manifest_volume(self):
        """
        If we restart but find remote volumes missing, we can easily end up
        with a manifest that lists "vol1, vol2, vol3, vol2", leaving a dangling
        vol3.  Make sure we can gracefully handle that.  This will only happen
        if the source data changes to be small enough to not create a vol3 on
        restart.
        """
        source = f"{_runtest_dir}/testfiles/largefiles"
        self.make_largefiles(count=5, size=1)
        self.backup("full", source, fail=3)
        # now delete the last volume on remote end and some source files
        assert not os.system(f"rm {_runtest_dir}/testfiles/output/duplicity-full*vol3.difftar*")
        assert not os.system(f"rm {source}/file[2345]")
        assert not os.system(f"echo hello > {source}/z")
        # finish backup
        self.backup("full", source)
        # and verify we can restore
        self.restore()

    def test_changed_source_file_disappears(self):
        """
        Make sure we correctly handle restarting a backup when a file
        disappears when we had been in the middle of backing it up.  It's
        possible that the first chunk of the next file will be skipped unless
        we're careful.
        """
        source = f"{_runtest_dir}/testfiles/largefiles"
        self.make_largefiles(count=1)
        self.backup("full", source, fail=2)
        # now remove starting source data and make sure we add something after
        assert not os.system(f"rm {source}/*")
        assert not os.system(f"echo hello > {source}/z")
        # finish backup
        self.backup("full", source)
        # and verify we can restore
        self.restore()
        assert not os.system(f"diff {source}/z {_runtest_dir}/testfiles/restore_out/z")


# Note that this class duplicates all the tests in RestartTest
class RestartTestWithoutEncryption(RestartTest):
    def setUp(self):
        super().setUp()
        self.class_args.extend(["--no-encryption"])

    def test_no_write_double_snapshot(self):
        """
        Test that restarting a full backup does not write duplicate entries
        into the sigtar, causing problems reading it back in older
        versions.
        https://launchpad.net/bugs/929067
        """
        self.make_largefiles()
        self.backup("full", f"{_runtest_dir}/testfiles/largefiles", fail=2)
        self.backup("full", f"{_runtest_dir}/testfiles/largefiles")
        # Now check sigtar
        sigtars = glob.glob(f"{_runtest_dir}/testfiles/output/duplicity-full*.sigtar.gz")
        self.assertEqual(1, len(sigtars))
        sigtar = sigtars[0]
        output = subprocess.Popen(["tar", "t", f"--file={sigtar}"], stdout=subprocess.PIPE).communicate()[0]
        self.assertEqual(1, output.split(b"\n").count(b"snapshot/"))

    def test_ignore_double_snapshot(self):
        """
        Test that we gracefully ignore double snapshot entries in a signature
        file.  This winds its way through duplicity as a deleted base dir,
        which doesn't make sense and should be ignored.  An older version of
        duplicity accidentally created such files as a result of a restart.
        https://launchpad.net/bugs/929067
        """

        if platform.system().startswith("Linux"):
            tarcmd = "tar"
        elif platform.system().startswith("Darwin"):
            tarcmd = "gtar"
        elif platform.system().endswith("BSD"):
            tarcmd = "gtar"
        else:
            raise Exception(f"Platform {platform.platform()} not supported by tar/gtar.")

        # Intial normal backup
        self.backup("full", f"{_runtest_dir}/testfiles/blocktartest")
        # Create an exact clone of the snapshot folder in the sigtar already.
        # Permissions and mtime must match.
        os.mkdir(f"{_runtest_dir}/testfiles/snapshot", 0o755)
        os.utime(f"{_runtest_dir}/testfiles/snapshot", (1030384548, 1030384548))
        # Adjust the sigtar.gz file to have a bogus second snapshot/ entry
        # at the beginning.
        sigtars = glob.glob(f"{_runtest_dir}/testfiles/output/duplicity-full*.sigtar.gz")
        self.assertEqual(1, len(sigtars))
        sigtar = sigtars[0]
        self.assertEqual(
            0,
            os.system(
                f"{tarcmd} c --file={_runtest_dir}/testfiles/snapshot.sigtar " f"-C {_runtest_dir}/testfiles snapshot"
            ),
        )
        self.assertEqual(0, os.system(f"gunzip -c {sigtar} > {_runtest_dir}/testfiles/full.sigtar"))
        self.assertEqual(
            0,
            os.system(
                f"{tarcmd} A --file={_runtest_dir}/testfiles/snapshot.sigtar " f"{_runtest_dir}/testfiles/full.sigtar"
            ),
        )
        self.assertEqual(0, os.system(f"gzip {_runtest_dir}/testfiles/snapshot.sigtar"))
        os.remove(sigtar)
        os.rename(f"{_runtest_dir}/testfiles/snapshot.sigtar.gz", sigtar)
        # Clear cache so our adjusted sigtar will be sync'd back into the cache
        self.assertEqual(0, os.system(f"rm -r {_runtest_dir}/testfiles/cache"))
        # Try a follow on incremental (which in buggy versions, would create
        # a deleted entry for the base dir)
        self.backup("inc", f"{_runtest_dir}/testfiles/blocktartest")
        self.assertEqual(
            1,
            len(glob.glob(f"{_runtest_dir}/testfiles/output/duplicity-new*.sigtar.gz")),
        )
        # Confirm we can restore it (which in buggy versions, would fail)
        self.restore()


class RestartTestConcurrent(RestartTest):
    def setUp(self):
        super().setUp()
        self.class_args.extend(["--concurrency=4"])


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