])

def test_pipreqcheck_virtualenv(host, virtualenv_activate_path):

    """

    Tests creation of Python virtual environment used for performing pip

    requirements upgrade checks.

    """

    with host.sudo():

        virtualenv_activate = host.file(virtualenv_activate_path)

        assert virtualenv_activate.is_file

        assert virtualenv_activate.user == 'pipreqcheck'

        assert virtualenv_activate.group == 'pipreqcheck'

        # @TODO: Possibly due to some timing issues, this file might

        # sometimes end-up being 0640, sometimes 0644.

        # assert virtualenv_activate.mode == 0o644

@pytest.mark.parametrize('config_dir', [

    '/etc/pip_check_requirements_upgrades',

    '/etc/pip_check_requirements_upgrades-py3',

])

def test_pipreqcheck_directories(host, config_dir):

    """

    Tests creation of directories used for storing configuration used by script

    that performs pip requirements upgrade checks.

    """

    with host.sudo():

        pipreqcheck_config_directory = host.file(config_dir)

        assert pipreqcheck_config_directory.is_directory

        assert pipreqcheck_config_directory.user == 'root'

        assert pipreqcheck_config_directory.group == 'pipreqcheck'

        assert pipreqcheck_config_directory.mode == 0o750

        pipreqcheck_config_directory_pipreqcheck = host.file(os.path.join(config_dir, 'pipreqcheck'))

        assert pipreqcheck_config_directory_pipreqcheck.is_directory

        assert pipreqcheck_config_directory_pipreqcheck.user == 'root'

        assert pipreqcheck_config_directory_pipreqcheck.group == 'pipreqcheck'

        assert pipreqcheck_config_directory_pipreqcheck.mode == 0o750

@pytest.mark.parametrize('requirements_in_path, requirements_txt_path', [

    ('/etc/pip_check_requirements_upgrades/pipreqcheck/requirements.in',

     '/etc/pip_check_requirements_upgrades/pipreqcheck/requirements.txt'),

    ('/etc/pip_check_requirements_upgrades-py3/pipreqcheck/requirements.in',

     '/etc/pip_check_requirements_upgrades-py3/pipreqcheck/requirements.txt'),

])

def test_pipreqcheck_requirements(host, requirements_in_path, requirements_txt_path):

    """

    Tests deployment of requirements input and text file used for virtual

    environment utilised by script that perform pip requirements upgrade checks.

    """

    with host.sudo():

        requirements_in = host.file(requirements_in_path)

        assert requirements_in.is_file

        assert requirements_in.user == 'root'

        assert requirements_in.group == 'pipreqcheck'

        assert requirements_in.mode == 0o640

        requirements_txt = host.file(requirements_txt_path)

        requirements_txt.is_file

        assert requirements_txt.user == 'root'

        assert requirements_txt.group == 'pipreqcheck'

        assert requirements_txt.mode == 0o640

@pytest.mark.parametrize("pip_path, expected_packages",  [

    ('/var/lib/pipreqcheck/virtualenv/bin/pip', [

        "Click==7.0",

        "pip==19.2.3",

        "pip-tools==4.0.0",

        "setuptools==41.2.0",

        "six==1.12.0",

        "wheel==0.33.6",

    ]),

    ('/var/lib/pipreqcheck/virtualenv-py3/bin/pip', [

        "Click==7.0",

        "pip==19.1.1",

        "pip-tools==3.9.0",

        "setuptools==41.2.0",

        "six==1.12.0",

        "wheel==0.33.6",

    ]),

])

def test_pipreqcheck_virtualenv_packages(host, pip_path, expected_packages):

    """

    Tests if correct packages are installed in virtualenv used for pip

    requirements checks..

    """

    packages = host.run("sudo -u %s %s freeze --all" % ('pipreqcheck', pip_path))

    # Normalise package names and order.

    expected_packages = sorted([unicode(p.lower()) for p in expected_packages])

    # This is a dummy distro-provided package ignored by the pip-tools.

    if "pkg-resources==0.0.0" in actual_packages:

        actual_packages.remove("pkg-resources==0.0.0")

    assert actual_packages == expected_packages

def test_pipreqcheck_script(host):

    """

    Tests script used for performing pip requirements upgrade checks.

    """

    pipreqcheck_script = host.file('/usr/local/bin/pip_check_requirements_upgrades.sh')

    assert pipreqcheck_script.is_file

    assert pipreqcheck_script.user == 'root'

    assert pipreqcheck_script.group == 'root'

    assert pipreqcheck_script.mode == 0o755

@pytest.mark.parametrize('crontab_path, virtualenv_path', [

    ('/etc/cron.d/check_pip_requirements', '/var/lib/pipreqcheck/virtualenv'),

    ('/etc/cron.d/check_pip_requirements-py3', '/var/lib/pipreqcheck/virtualenv-py3'),

])

def test_pipreqcheck_crontab(host, crontab_path, virtualenv_path):

    """

    Tests if crontab entry is set-up correctly for running the pip requirements

    upgrade checks.

    """

    crontab = host.file(crontab_path)

    assert crontab.is_file

    assert crontab.user == 'root'

    assert crontab.group == 'root'

    assert crontab.mode == 0o644

    assert "MAILTO=root" in crontab.content

    assert virtualenv_path in crontab.content.split(" ")

@pytest.mark.parametrize('python_path, expected_major_version', [

    ('/var/lib/pipreqcheck/virtualenv/bin/python', '2'),

    ('/var/lib/pipreqcheck/virtualenv-py3/bin/python', '3'),

])

def test_pipreqcheck_virtualenv_python_version(host, python_path, expected_major_version):

    """

    Tests if Python virtual environment for pipreqcheck has been

    set-up correctly.

    """

    with host.sudo('pipreqcheck'):

        major_version = host.run("%s -c %s", python_path, "import sys; print(sys.version_info.major)")

    assert major_version.rc == 0

@pytest.mark.parametrize('wrong_python_path', [

    '/var/lib/pipreqcheck/virtualenv/bin/python3',

    '/var/lib/pipreqcheck/virtualenv-py3/bin/python2',

])

def test_pipreqcheck_virtualenv_wrong_python_version_not_present(host, wrong_python_path):

    """

    Tests if wrong version of Python 2 is absent or not.

    """

    with host.sudo():

        wrong_python_path_file = host.file(wrong_python_path)

        assert not wrong_python_path_file.exists

        ip6tables = host.command('ip6tables-save')

        assert ip6tables.rc == 0

        assert "-A flood -p icmp -m icmp --icmp-type 8 -m hashlimit --hashlimit-upto 5/sec --hashlimit-burst 5 " \

            "--hashlimit-mode srcip --hashlimit-name icmp -j RETURN" in iptables.stdout

        assert "-A flood -p ipv6-icmp -m icmp6 --icmpv6-type 128 -m hashlimit --hashlimit-upto 5/sec --hashlimit-burst 5 " \

            "--hashlimit-mode srcip --hashlimit-name icmp -j RETURN" in ip6tables.stdout

def test_pipreqcheck_virtualenv_user(host):

    """

    Tests if group and user for running pip requirements upgrade checks have

    been created correctly with user-provided uid/gid.

    """

    group = host.group('pipreqcheck')

    assert group.exists

    assert group.gid == 2500

    user = host.user('pipreqcheck')

    assert user.exists

    assert user.home == '/var/lib/pipreqcheck'

    assert user.uid == 2500

    assert user.group == 'pipreqcheck'

    assert user.groups == ['pipreqcheck']

def test_backup_configuration(host):

    """

    Tests if backup configuration has been deployed correctly.

    """

    with host.sudo():

        common = host.file('/etc/duply/main/patterns/common')

        assert common.is_file

        assert "/var/log" in common.content.split("\n")

        assert "/etc/shadow" in common.content.split("\n")

        assert "/var/mail" in common.content.split("\n")

        assert "/var/spool/cron" in common.content.split("\n")

        common_extra = host.file('/etc/duply/main/patterns/common_extra')

        assert common_extra.is_file

        assert "/home/user1" in common_extra.content.split("\n")

        assert "/home/user2" in common_extra.content.split("\n")

def test_ntp_software_installed(host):

    """

    Tests if NTP packages are installed.

    """

    assert host.package('ntp').is_installed

    assert host.package('ntpdate').is_installed

def test_ntp_server_configuration(host):

    """

    Tests if NTP server has been correctly configured.

    """

    with host.sudo():

        # Read the configuration file.

        configuration = host.file("/etc/ntp.conf").content.split("\n")

        # Extract only the relevant sections of files (exculde empty

        # lines and comments).

        configuration = [c.strip() for c in configuration if re.match(r'^\s*(|#.*)$', c) is None]

        # Ensure correct servers have been configured in the pool.

        servers = [c for c in configuration if c.startswith('server')]

        expected_servers = ["server 0.debian.pool.ntp.org iburst",

                            "server 1.debian.pool.ntp.org iburst",

                            "server 2.debian.pool.ntp.org iburst"]

        assert sorted(servers) == sorted(expected_servers)

        # Ensure querying of server is disable for untrusted clients.

        restrictions = [c for c in configuration if c.startswith('restrict')]

        expected_restrictions = ["restrict -4 default kod notrap nomodify nopeer noquery",

                                 "restrict -6 default kod notrap nomodify nopeer noquery",

                                 "restrict 127.0.0.1",

                                 "restrict ::1"]

        assert sorted(restrictions) == sorted(expected_restrictions)

def test_ntp_query_server_count(host):

    # Two lines for headers, and one line per configured server.

    expected_stdout_line_count = 5

    ntpq = host.command("ntpq -p -n")

    assert ntpq.rc == 0

def test_ntp_listening_interfaces(host):

    """

    Tests if NTP server is listening on correct ports.

    """

    assert host.socket('udp://:::123').is_listening