# coding: utf-8
# PyDERASN -- Python ASN.1 DER/CER/BER codec with abstract structures
# Copyright (C) 2017-2020 Sergey Matveev <stargrave@stargrave.org>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as
# published by the Free Software Foundation, version 3 of the License.
#
# This program 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 Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this program.  If not, see
# <http://www.gnu.org/licenses/>.
"""CRL related schemas, just to test the performance with them
"""

from os.path import exists
from time import time
from unittest import skipIf
from unittest import TestCase

from six import PY2

from pyderasn import BitString
from pyderasn import encode_cer
from pyderasn import file_mmaped
from pyderasn import Sequence
from pyderasn import SequenceOf
from pyderasn import tag_ctxc

from tests.test_crts import AlgorithmIdentifier
from tests.test_crts import CertificateSerialNumber
from tests.test_crts import Extensions
from tests.test_crts import Name
from tests.test_crts import Time
from tests.test_crts import Version


class RevokedCertificate(Sequence):
    schema = (
        ("userCertificate", CertificateSerialNumber()),
        ("revocationDate", Time()),
        ("crlEntryExtensions", Extensions(optional=True)),
    )


class RevokedCertificates(SequenceOf):
    schema = RevokedCertificate()


class TBSCertList(Sequence):
    schema = (
        ("version", Version(optional=True)),
        ("signature", AlgorithmIdentifier()),
        ("issuer", Name()),
        ("thisUpdate", Time()),
        ("nextUpdate", Time(optional=True)),
        ("revokedCertificates", RevokedCertificates(optional=True)),
        ("crlExtensions", Extensions(expl=tag_ctxc(0), optional=True)),
    )


class CertificateList(Sequence):
    schema = (
        ("tbsCertList", TBSCertList()),
        ("signatureAlgorithm", AlgorithmIdentifier()),
        ("signatureValue", BitString()),
    )


CRL_PATH = "revoke.crl"


@skipIf(not exists(CRL_PATH), "CACert's revoke.crl not found")
class TestCACert(TestCase):
    def test_cer(self):
        with open(CRL_PATH, "rb") as fd:
            raw = fd.read()
        print("DER read")
        start = time()
        crl1 = CertificateList().decod(raw)
        print("DER decoded", time() - start)
        start = time()
        cer_raw = encode_cer(crl1)
        print("CER encoded", time() - start)
        start = time()
        crl2 = CertificateList().decod(cer_raw, ctx={"bered": True})
        print("CER decoded", time() - start)
        self.assertEqual(crl2, crl1)
        start = time()
        der_raw = crl2.encode()
        print("DER encoded", time() - start)
        self.assertSequenceEqual(der_raw, raw)

    @skipIf(PY2, "Py27 mmap does not implement buffer protocol")
    def test_mmaped(self):
        fd = open(CRL_PATH, "rb")
        start = time()
        CertificateList().decod(file_mmaped(fd))
        print("DER decoded", time() - start)

    def test_evgens(self):
        fd = open(CRL_PATH, "rb")
        raw = memoryview(fd.read()) if PY2 else file_mmaped(fd)
        print("CRL opened")
        evgens_count = 0
        revoked_certs_count = 0
        start = time()
        for decode_path, _, _ in CertificateList().decode_evgen(raw):
            evgens_count += 1
            if (
                    len(decode_path) == 3 and
                    decode_path[:2] == ("tbsCertList", "revokedCertificates")
            ):
                revoked_certs_count += 1
        print("CRL parsed", time() - start)
        evgens_upto_count = 0
        start = time()
        for decode_path, _, _ in CertificateList().decode_evgen(raw, ctx={
                "evgen_mode_upto": (
                    (("tbsCertList", "revokedCertificates", any), True),
                ),
        }):
            evgens_upto_count += 1
        print("CRL upto parsed", time() - start)
        self.assertEqual(
            float(evgens_count - evgens_upto_count) / revoked_certs_count,
            3,
        )