import os

from fixtures import TempDir

from testtools import ExpectedException, TestCase, run_test_with
from testtools.assertions import assert_that
from testtools.matchers import (
    Contains, DirExists, Equals, FileContains, FileExists, MatchesStructure)
from testtools.twistedsupport import (
    AsynchronousDeferredRunTest, flush_logged_errors)

from twisted.internet import reactor
from twisted.internet.defer import inlineCallbacks
from twisted.internet.error import CannotListenError, ConnectionRefusedError

from txacme.urls import LETSENCRYPT_STAGING_DIRECTORY

from marathon_acme.cli import init_storage_dir, main, parse_listen_addr


# Make sure we always use the Let's Encrypt Staging endpoint for these tests
def main_t(reactor, **kwargs):
    argv = kwargs.get('argv', [])
    env = kwargs.get('env', {})
    return main(reactor, acme_url=LETSENCRYPT_STAGING_DIRECTORY.asText(),
                argv=argv, env=env)


class TestCli(TestCase):
    # These are testtools-style tests so we can run aynchronous tests

    def test_storage_dir_required(self):
        """
        When the program is run with no arguments, it should exit with code 2
        because there is one required argument.
        """
        with ExpectedException(SystemExit, MatchesStructure(code=Equals(2))):
            main_t(reactor, argv=[])

    @inlineCallbacks
    @run_test_with(AsynchronousDeferredRunTest.make_factory(timeout=10.0))
    def test_storage_dir_provided(self):
        """
        When the program is run with an argument, it should start up and run.
        The program is expected to fail because it is unable to connect to
        Marathon.

        This test takes a while because we have to let txacme go through it's
        initial sync (registration + issuing of 0 certificates) before things
        can be halted.
        """
        temp_dir = self.useFixture(TempDir())
        yield main_t(reactor, argv=[
            temp_dir.path,
            '--acme', LETSENCRYPT_STAGING_DIRECTORY.asText(),
            '--marathon', 'http://localhost:28080'  # An address we can't reach
        ])

        # Expect a 'certs' directory to be created
        self.assertThat(os.path.isdir(temp_dir.join('certs')), Equals(True))

        # Expect an 'unmanaged-certs' directory to be created
        self.assertThat(
            os.path.isdir(temp_dir.join('unmanaged-certs')), Equals(True))

        # Expect a default certificate to be created
        self.assertThat(os.path.isfile(temp_dir.join('default.pem')),
                        Equals(True))

        # Expect to be unable to connect to Marathon
        flush_logged_errors(ConnectionRefusedError)

    @inlineCallbacks
    @run_test_with(AsynchronousDeferredRunTest.make_factory(timeout=5.0))
    def test_storage_dir_provided_vault(self):
        """
        When the program is run with an argument and the --vault option, it
        should start up and run. The program is expected to fail because it is
        unable to connect to Vault.

        Unlike the above test, this crashes immediately and returns because we
        never actually start up txacme or marathon-acme if we can't get/store
        an ACME client key.
        """
        with ExpectedException(ConnectionRefusedError,
                               r'Connection was refused by other side'):
            yield main_t(
                reactor,
                env={
                    # An address we can't reach
                    'VAULT_ADDR': 'http://localhost:28080'
                },
                argv=[
                    'secret',
                    '--vault',
                    '--acme', LETSENCRYPT_STAGING_DIRECTORY.asText(),
                ]
            )

        flush_logged_errors(ConnectionRefusedError)

    @inlineCallbacks
    @run_test_with(AsynchronousDeferredRunTest.make_factory(timeout=5.0))
    def test_cannot_listen(self):
        """
        When the program is run with an argument and a listen address specified
        with an address that we can't listen on (e.g. 1.1.1.1), a
        CannotListenError is expected to be logged and the program should stop.
        """
        temp_dir = self.useFixture(TempDir())
        yield main_t(reactor, argv=[
            temp_dir.path,
            '--listen', '1.1.1.1:8080',  # An address we can't listen on
        ])

        # Expect a 'certs' directory to be created
        self.assertThat(os.path.isdir(temp_dir.join('certs')), Equals(True))

        # Expect a default certificate to be created
        self.assertThat(os.path.isfile(temp_dir.join('default.pem')),
                        Equals(True))

        # Expect to be unable to listen
        flush_logged_errors(CannotListenError)


class TestParseListenAddr(object):
    def test_parse_no_colon(self):
        """
        When a listen address is parsed with no ':' character, an error is
        raised.
        """
        with ExpectedException(
            ValueError,
            r"'foobar' does not have the correct form for a listen address: "
                r'\[ipaddress\]:port'):
            parse_listen_addr('foobar')

    def test_parse_no_ip_address(self):
        """
        When a listen address is parsed with no IP address, an endpoint
        description with the listen address's port but no interface is
        returned.
        """
        assert_that(parse_listen_addr(':8080'), Equals('tcp:8080'))

    def test_parse_ipv4(self):
        """
        When a listen address is parsed with an IPv4 address, an appropriate
        interface is present in the returned endpoint description.
        """
        assert_that(parse_listen_addr('127.0.0.1:8080'),
                    Equals('tcp:8080:interface=127.0.0.1'))

    def test_parse_ipv6(self):
        """
        When a listen address is parsed with an IPv4 address, an appropriate
        interface is present in the returned endpoint description.
        """
        assert_that(parse_listen_addr('[::]:8080'),
                    Equals('tcp6:8080:interface=\\:\\:'))

    def test_parse_invalid_ipaddress(self):
        """
        When a listen address is parsed with an invalid IP address, an error
        is raised.
        """
        with ExpectedException(
            ValueError,
                r"u?'hello' does not appear to be an IPv4 or IPv6 address"):
            parse_listen_addr('hello:8080')

    def test_parse_invalid_port(self):
        """
        When a listen address is parsed with an invalid port, an error is
        raised.
        """
        with ExpectedException(
                ValueError,
                r"'foo' does not appear to be a valid port number"):
            parse_listen_addr(':foo')

        with ExpectedException(
                ValueError,
                r"'0' does not appear to be a valid port number"):
            parse_listen_addr(':0')

        with ExpectedException(
                ValueError,
                r"'65536' does not appear to be a valid port number"):
            parse_listen_addr(':65536')

        with ExpectedException(
                ValueError,
                r"'' does not appear to be a valid port number"):
            parse_listen_addr(':')


class TestInitStorageDir(object):
    def test_files_created_if_not_exist(self, tmpdir):
        """
        When the certificate directory does not contain a 'default.pem' file
        and a 'certs' directory, calling init_storage_dir() should create a
        'default.pem' file with x509 certificate data and create a 'certs'
        directory.
        """
        init_storage_dir(str(tmpdir))

        assert_that(str(tmpdir.join('default.pem')), FileExists())
        # Check that this *looks* like a x509 cert
        assert_that(str(tmpdir.join('default.pem')),
                    FileContains(matcher=Contains(
                        '-----BEGIN RSA PRIVATE KEY-----')))

        assert_that(str(tmpdir.join('certs')), DirExists())

    def test_files_not_created_if_exist(self, tmpdir):
        """
        When the certificate directory does contain a 'default.pem' file
        and a 'certs' directory, calling init_storage_dir() should not attempt
        to create those files.
        """
        tmpdir.join('default.pem').write('blah')
        tmpdir.join('certs').mkdir()
        tmpdir.join('unmanaged-certs').mkdir()

        init_storage_dir(str(tmpdir))

        assert_that(str(tmpdir.join('default.pem')), FileExists())
        # Check that the file hasn't changed
        assert_that(str(tmpdir.join('default.pem')), FileContains('blah'))

        assert_that(str(tmpdir.join('certs')), DirExists())

import os

from fixtures import TempDir

from testtools import ExpectedException, TestCase, run_test_with
from testtools.assertions import assert_that
from testtools.matchers import (
    Contains, DirExists, Equals, FileContains, FileExists, MatchesStructure)
from testtools.twistedsupport import (
    AsynchronousDeferredRunTest, flush_logged_errors)

from twisted.internet import reactor
from twisted.internet.defer import inlineCallbacks
from twisted.internet.error import CannotListenError, ConnectionRefusedError

from txacme.urls import LETSENCRYPT_STAGING_DIRECTORY

from marathon_acme.cli import init_storage_dir, main, parse_listen_addr


# Make sure we always use the Let's Encrypt Staging endpoint for these tests
def main_t(reactor, **kwargs):
    argv = kwargs.get('argv', [])
    env = kwargs.get('env', {})
    return main(reactor, acme_url=LETSENCRYPT_STAGING_DIRECTORY.asText(),
                argv=argv, env=env)


class TestCli(TestCase):
    # These are testtools-style tests so we can run aynchronous tests

    def test_storage_dir_required(self):
        """
        When the program is run with no arguments, it should exit with code 2
        because there is one required argument.
        """
        with ExpectedException(SystemExit, MatchesStructure(code=Equals(2))):
            main_t(reactor, argv=[])

    @inlineCallbacks
    @run_test_with(AsynchronousDeferredRunTest.make_factory(timeout=10.0))
    def test_storage_dir_provided(self):
        """
        When the program is run with an argument, it should start up and run.
        The program is expected to fail because it is unable to connect to
        Marathon.

        This test takes a while because we have to let txacme go through it's
        initial sync (registration + issuing of 0 certificates) before things
        can be halted.
        """
        temp_dir = self.useFixture(TempDir())
        yield main_t(reactor, argv=[
            temp_dir.path,
            '--acme', LETSENCRYPT_STAGING_DIRECTORY.asText(),
            '--marathon', 'http://localhost:28080'  # An address we can't reach
        ])

        # Expect a 'certs' directory to be created
        self.assertThat(os.path.isdir(temp_dir.join('certs')), Equals(True))

        # Expect an 'unmanaged-certs' directory to be created
        self.assertThat(
            os.path.isdir(temp_dir.join('unmanaged-certs')), Equals(True))

        # Expect a default certificate to be created
        self.assertThat(os.path.isfile(temp_dir.join('default.pem')),
                        Equals(True))

        # Expect to be unable to connect to Marathon
        flush_logged_errors(ConnectionRefusedError)

    @inlineCallbacks
    @run_test_with(AsynchronousDeferredRunTest.make_factory(timeout=5.0))
    def test_storage_dir_provided_vault(self):
        """
        When the program is run with an argument and the --vault option, it
        should start up and run. The program is expected to fail because it is
        unable to connect to Vault.

        Unlike the above test, this crashes immediately and returns because we
        never actually start up txacme or marathon-acme if we can't get/store
        an ACME client key.
        """
        with ExpectedException(ConnectionRefusedError,
                               r'Connection was refused by other side'):
            yield main_t(
                reactor,
                env={
                    # An address we can't reach
                    'VAULT_ADDR': 'http://localhost:28080'
                },
                argv=[
                    'secret',
                    '--vault',
                    '--acme', LETSENCRYPT_STAGING_DIRECTORY.asText(),
                ]
            )

        flush_logged_errors(ConnectionRefusedError)

    @inlineCallbacks
    @run_test_with(AsynchronousDeferredRunTest.make_factory(timeout=5.0))
    def test_cannot_listen(self):
        """
        When the program is run with an argument and a listen address specified
        with an address that we can't listen on (e.g. 1.1.1.1), a
        CannotListenError is expected to be logged and the program should stop.
        """
        temp_dir = self.useFixture(TempDir())
        yield main_t(reactor, argv=[
            temp_dir.path,
            '--listen', '1.1.1.1:8080',  # An address we can't listen on
        ])

        # Expect a 'certs' directory to be created
        self.assertThat(os.path.isdir(temp_dir.join('certs')), Equals(True))

        # Expect a default certificate to be created
        self.assertThat(os.path.isfile(temp_dir.join('default.pem')),
                        Equals(True))

        # Expect to be unable to listen
        flush_logged_errors(CannotListenError)


class TestParseListenAddr(object):
    def test_parse_no_colon(self):
        """
        When a listen address is parsed with no ':' character, an error is
        raised.
        """
        with ExpectedException(
            ValueError,
            r"'foobar' does not have the correct form for a listen address: "
                r'\[ipaddress\]:port'):
            parse_listen_addr('foobar')

    def test_parse_no_ip_address(self):
        """
        When a listen address is parsed with no IP address, an endpoint
        description with the listen address's port but no interface is
        returned.
        """
        assert_that(parse_listen_addr(':8080'), Equals('tcp:8080'))

    def test_parse_ipv4(self):
        """
        When a listen address is parsed with an IPv4 address, an appropriate
        interface is present in the returned endpoint description.
        """
        assert_that(parse_listen_addr('127.0.0.1:8080'),
                    Equals('tcp:8080:interface=127.0.0.1'))

    def test_parse_ipv6(self):
        """
        When a listen address is parsed with an IPv4 address, an appropriate
        interface is present in the returned endpoint description.
        """
        assert_that(parse_listen_addr('[::]:8080'),
                    Equals('tcp6:8080:interface=\\:\\:'))

    def test_parse_invalid_ipaddress(self):
        """
        When a listen address is parsed with an invalid IP address, an error
        is raised.
        """
        with ExpectedException(
            ValueError,
                r"u?'hello' does not appear to be an IPv4 or IPv6 address"):
            parse_listen_addr('hello:8080')

    def test_parse_invalid_port(self):
        """
        When a listen address is parsed with an invalid port, an error is
        raised.
        """
        with ExpectedException(
                ValueError,
                r"'foo' does not appear to be a valid port number"):
            parse_listen_addr(':foo')

        with ExpectedException(
                ValueError,
                r"'0' does not appear to be a valid port number"):
            parse_listen_addr(':0')

        with ExpectedException(
                ValueError,
                r"'65536' does not appear to be a valid port number"):
            parse_listen_addr(':65536')

        with ExpectedException(
                ValueError,
                r"'' does not appear to be a valid port number"):
            parse_listen_addr(':')


class TestInitStorageDir(object):
    def test_files_created_if_not_exist(self, tmpdir):
        """
        When the certificate directory does not contain a 'default.pem' file
        and a 'certs' directory, calling init_storage_dir() should create a
        'default.pem' file with x509 certificate data and create a 'certs'
        directory.
        """
        init_storage_dir(str(tmpdir))

        assert_that(str(tmpdir.join('default.pem')), FileExists())
        # Check that this *looks* like a x509 cert
        assert_that(str(tmpdir.join('default.pem')),
                    FileContains(matcher=Contains(
                        '-----BEGIN RSA PRIVATE KEY-----')))

        assert_that(str(tmpdir.join('certs')), DirExists())

    def test_files_not_created_if_exist(self, tmpdir):
        """
        When the certificate directory does contain a 'default.pem' file
        and a 'certs' directory, calling init_storage_dir() should not attempt
        to create those files.
        """
        tmpdir.join('default.pem').write('blah')
        tmpdir.join('certs').mkdir()
        tmpdir.join('unmanaged-certs').mkdir()

        init_storage_dir(str(tmpdir))

        assert_that(str(tmpdir.join('default.pem')), FileExists())
        # Check that the file hasn't changed
        assert_that(str(tmpdir.join('default.pem')), FileContains('blah'))

        assert_that(str(tmpdir.join('certs')), DirExists())

# Make sure we always use the Let's Encrypt Staging endpoint for these tests # These are testtools-style tests so we can run aynchronous tests When the program is run with no arguments, it should exit with code 2         because there is one required argument. When the program is run with an argument, it should start up and run.         The program is expected to fail because it is unable to connect to         Marathon.          This test takes a while because we have to let txacme go through it's         initial sync (registration + issuing of 0 certificates) before things         can be halted. # An address we can't reach # Expect a 'certs' directory to be created # Expect an 'unmanaged-certs' directory to be created # Expect a default certificate to be created # Expect to be unable to connect to Marathon When the program is run with an argument and the --vault option, it         should start up and run. The program is expected to fail because it is         unable to connect to Vault.          Unlike the above test, this crashes immediately and returns because we         never actually start up txacme or marathon-acme if we can't get/store         an ACME client key. # An address we can't reach When the program is run with an argument and a listen address specified         with an address that we can't listen on (e.g. 1.1.1.1), a         CannotListenError is expected to be logged and the program should stop. # An address we can't listen on # Expect a 'certs' directory to be created # Expect a default certificate to be created # Expect to be unable to listen When a listen address is parsed with no ':' character, an error is         raised. When a listen address is parsed with no IP address, an endpoint         description with the listen address's port but no interface is         returned. When a listen address is parsed with an IPv4 address, an appropriate         interface is present in the returned endpoint description. When a listen address is parsed with an IPv4 address, an appropriate         interface is present in the returned endpoint description. When a listen address is parsed with an invalid IP address, an error         is raised. When a listen address is parsed with an invalid port, an error is         raised. When the certificate directory does not contain a 'default.pem' file         and a 'certs' directory, calling init_storage_dir() should create a         'default.pem' file with x509 certificate data and create a 'certs'         directory. # Check that this *looks* like a x509 cert When the certificate directory does contain a 'default.pem' file         and a 'certs' directory, calling init_storage_dir() should not attempt         to create those files. # Check that the file hasn't changed