# -*- coding: utf-8 -*-
# BSD 2-Clause License
#
# Apprise - Push Notification Library.
# Copyright (c) 2025, Chris Caron <lead2gold@gmail.com>
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright notice,
#    this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright notice,
#    this list of conditions and the following disclaimer in the documentation
#    and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.

import logging
import os
import sys
import pytest
from unittest import mock

from apprise import AppriseAsset
from apprise import PersistentStoreMode
from apprise import utils

# Disable logging for a cleaner testing output
logging.disable(logging.CRITICAL)

# Attachment Directory
TEST_VAR_DIR = os.path.join(os.path.dirname(__file__), 'var')


@pytest.mark.skipif(
    'cryptography' not in sys.modules, reason="Requires cryptography")
def test_utils_pem_general(tmpdir):
    """
    Utils:PEM

    """

    # string to manipulate/work with
    unencrypted_str = "message"

    tmpdir0 = tmpdir.mkdir('tmp00')

    # Currently no files here
    assert os.listdir(str(tmpdir0)) == []

    asset = AppriseAsset(
        storage_mode=PersistentStoreMode.MEMORY,
        storage_path=str(tmpdir0),
        pem_autogen=False,
    )

    # Create a PEM Controller
    pem_c = utils.pem.ApprisePEMController(path=None, asset=asset)

    # Nothing to lookup
    assert pem_c.public_keyfile() is None
    assert pem_c.public_key() is None
    assert pem_c.x962_str == ''
    assert pem_c.decrypt(b'data') is None
    assert pem_c.encrypt(unencrypted_str) is None
    # Keys can not be generated in memory mode
    assert pem_c.keygen() is False
    assert pem_c.sign(b'data') is None

    asset = AppriseAsset(
        storage_mode=PersistentStoreMode.FLUSH,
        storage_path=str(tmpdir0),
        pem_autogen=False,
    )

    # No new files
    assert os.listdir(str(tmpdir0)) == []

    # Our asset is now write mode, so we will be able to generate a key
    pem_c = utils.pem.ApprisePEMController(path=str(tmpdir0), asset=asset)
    # Nothing to lookup
    assert pem_c.public_keyfile() is None
    assert pem_c.public_key() is None
    assert pem_c.x962_str == ''
    assert pem_c.encrypt(unencrypted_str) is None

    # Generate our keys
    assert bool(pem_c) is False
    assert pem_c.keygen() is True
    assert bool(pem_c) is True

    # We have 2 new key files generated
    pub_keyfile = os.path.join(str(tmpdir0), 'public_key.pem')
    prv_keyfile = os.path.join(str(tmpdir0), 'private_key.pem')
    assert os.path.isfile(pub_keyfile)
    assert os.path.isfile(prv_keyfile)
    assert pem_c.public_keyfile() is not None
    assert pem_c.decrypt("garbage") is None
    assert pem_c.public_key() is not None

    # Keys used later on as ref
    pubkey_ref = pem_c.public_key()
    prvkey_ref = pem_c.private_key()

    assert isinstance(pem_c.x962_str, str)
    assert len(pem_c.x962_str) > 20
    content = pem_c.encrypt(unencrypted_str)
    assert pem_c.decrypt(pem_c.encrypt(unencrypted_str.encode('utf-8'))) \
        == pem_c.decrypt(pem_c.encrypt(unencrypted_str))
    assert pem_c.decrypt(
        pem_c.encrypt(unencrypted_str, public_key=pem_c.public_key())) \
        == pem_c.decrypt(pem_c.encrypt(unencrypted_str))
    assert pem_c.decrypt(content) == unencrypted_str
    assert isinstance(content, str)
    assert pem_c.decrypt(content) == unencrypted_str
    # support str as well
    assert pem_c.decrypt(content) == unencrypted_str
    assert pem_c.decrypt(content.encode('utf-8')) == unencrypted_str
    # Sign test
    assert isinstance(pem_c.sign(content.encode('utf-8')), bytes)

    # Web Push handling
    webpush_content = pem_c.encrypt_webpush(
        unencrypted_str,
        public_key=pem_c.public_key(),
        auth_secret=b'secret')
    assert isinstance(webpush_content, bytes)

    webpush_content = pem_c.encrypt_webpush(
        unencrypted_str.encode('utf-8'),
        public_key=pem_c.public_key(),
        auth_secret=b'secret')
    assert isinstance(webpush_content, bytes)

    # Non Bytes (garbage basically)
    with pytest.raises(TypeError):
        assert pem_c.decrypt(None) is None

    with pytest.raises(TypeError):
        assert pem_c.decrypt(5) is None

    with pytest.raises(TypeError):
        assert pem_c.decrypt(False) is None

    with pytest.raises(TypeError):
        assert pem_c.decrypt(object) is None

    # Test our initialization
    pem_c = utils.pem.ApprisePEMController(
        path=None,
        prv_keyfile='invalid',
        asset=asset)
    assert pem_c.private_keyfile() is False
    assert pem_c.public_keyfile() is None
    assert pem_c.prv_keyfile is False
    assert pem_c.pub_keyfile is None
    assert pem_c.private_key() is None
    assert pem_c.public_key() is None
    assert pem_c.decrypt(content) is None

    pem_c = utils.pem.ApprisePEMController(
        path=None,
        pub_keyfile='invalid',
        asset=asset)
    assert pem_c.private_keyfile() is None
    assert pem_c.public_keyfile() is False
    assert pem_c.prv_keyfile is None
    assert pem_c.pub_keyfile is False
    assert pem_c.private_key() is None
    assert pem_c.public_key() is None
    assert pem_c.decrypt(content) is None

    pem_c = utils.pem.ApprisePEMController(
        path=None,
        prv_keyfile=prv_keyfile,
        asset=asset)
    assert pem_c.private_keyfile() == prv_keyfile
    assert pem_c.public_keyfile() is None
    assert pem_c.private_key() is not None
    assert pem_c.prv_keyfile == prv_keyfile
    assert pem_c.pub_keyfile is None
    assert pem_c.public_key() is not None
    assert pem_c.decrypt(content) == unencrypted_str

    pem_c = utils.pem.ApprisePEMController(
        path=None,
        pub_keyfile=pub_keyfile,
        asset=asset)
    assert pem_c.private_keyfile() is None
    assert pem_c.public_keyfile() == pub_keyfile
    assert pem_c.prv_keyfile is None
    assert pem_c.pub_keyfile == pub_keyfile
    assert pem_c.private_key() is None
    assert pem_c.public_key() is not None
    assert pem_c.decrypt(content) is None

    # Test our path references
    pem_c = utils.pem.ApprisePEMController(path=str(tmpdir0), asset=asset)
    assert pem_c.load_private_key(path=None) is True
    assert pem_c.private_keyfile() == prv_keyfile
    assert pem_c.prv_keyfile is None
    assert pem_c.pub_keyfile is None
    assert pem_c.decrypt(content) == unencrypted_str

    # Generate a new key referencing another location
    pem_c = utils.pem.ApprisePEMController(
        name='keygen-tests',
        path=str(tmpdir0), asset=asset)

    # generate ourselves some keys
    assert pem_c.keygen() is True
    keygen_prv_file = pem_c.prv_keyfile
    keygen_pub_file = pem_c.pub_keyfile

    # Remove 1 (but not both)
    os.unlink(keygen_pub_file)

    pem_c = utils.pem.ApprisePEMController(
        name='keygen-tests',
        path=str(tmpdir0), asset=asset)
    # Private key was found, so this does not work
    assert pem_c.keygen() is False
    os.unlink(keygen_prv_file)

    pem_c = utils.pem.ApprisePEMController(
        name='keygen-tests',
        path=str(tmpdir0), asset=asset)
    # It works now
    assert pem_c.keygen() is True

    # Tests public_key generation failure only
    with mock.patch('builtins.open', side_effect=OSError()):
        assert pem_c.keygen(force=True) is False
        with mock.patch('os.unlink', side_effect=OSError()):
            assert pem_c.keygen(force=True) is False
        with mock.patch('os.unlink', return_value=True):
            assert pem_c.keygen(force=True) is False

    # Tests private key generation
    side_effect = [
        mock.mock_open(read_data="file contents").return_value] + \
        [OSError() for _ in range(10)]
    with mock.patch('builtins.open', side_effect=side_effect):
        assert pem_c.keygen(force=True) is False
    with mock.patch('builtins.open', side_effect=side_effect):
        with mock.patch('os.unlink', side_effect=OSError()):
            assert pem_c.keygen(force=True) is False
    with mock.patch('builtins.open', side_effect=side_effect):
        with mock.patch('os.unlink', return_value=True):
            assert pem_c.keygen(force=True) is False

    # Generate a new key referencing another location
    pem_c = utils.pem.ApprisePEMController(path=str(tmpdir0), asset=asset)
    # We can't re-generate keys if ones already exist
    assert pem_c.keygen() is False
    # the keygen is the big difference here
    assert pem_c.keygen(name='test') is True
    # under the hood, a key is not regenerated (as one already exists)
    assert pem_c.keygen(name='test') is False
    # Generate it a second time by force
    assert pem_c.keygen(name='test', force=True) is True

    assert pem_c.private_keyfile() == os.path.join(
        str(tmpdir0), 'test-private_key.pem')
    assert pem_c.public_keyfile() == os.path.join(
        str(tmpdir0), 'test-public_key.pem')
    assert pem_c.private_key() is not None
    assert pem_c.public_key() is not None
    assert pem_c.prv_keyfile == os.path.join(
        str(tmpdir0), 'test-private_key.pem')
    assert pem_c.pub_keyfile == os.path.join(
        str(tmpdir0), 'test-public_key.pem')
    # 'content' was generated using a different key and can not be
    # decrypted
    assert pem_c.decrypt(content) is None

    # Test Decryption files
    pem_c = utils.pem.ApprisePEMController(path=str(tmpdir0), asset=asset)
    # Calling decrypt triggers underlining code to auto-load
    assert pem_c.decrypt(content) == unencrypted_str
    # Using a private key by path
    assert pem_c.decrypt(
        content, private_key=pem_c.private_key()) == unencrypted_str

    # Test different edge cases of load_private_key()
    pem_c = utils.pem.ApprisePEMController(path=str(tmpdir0), asset=asset)
    assert pem_c.load_private_key() is True
    pem_c = utils.pem.ApprisePEMController(path=str(tmpdir0), asset=asset)
    assert pem_c.load_private_key(path=prv_keyfile) is True
    pem_c = utils.pem.ApprisePEMController(path=str(tmpdir0), asset=asset)
    with mock.patch('builtins.open', side_effect=TypeError()):
        assert pem_c.load_private_key(path=prv_keyfile) is False
    with mock.patch('builtins.open', side_effect=OSError()):
        assert pem_c.load_private_key(path=prv_keyfile) is False
    with mock.patch('builtins.open', side_effect=FileNotFoundError()):
        assert pem_c.load_private_key(path=prv_keyfile) is False

    # Test different edge cases of load_public_key()
    pem_c = utils.pem.ApprisePEMController(path=str(tmpdir0), asset=asset)
    assert pem_c.load_public_key() is True
    pem_c = utils.pem.ApprisePEMController(path=str(tmpdir0), asset=asset)
    assert pem_c.load_public_key(path=pub_keyfile) is True
    pem_c = utils.pem.ApprisePEMController(path=str(tmpdir0), asset=asset)
    with mock.patch('builtins.open', side_effect=TypeError()):
        assert pem_c.load_public_key(path=pub_keyfile) is False
    with mock.patch('builtins.open', side_effect=OSError()):
        assert pem_c.load_public_key(path=pub_keyfile) is False
    with mock.patch('builtins.open', side_effect=FileNotFoundError()):
        assert pem_c.load_public_key(path=pub_keyfile) is False

    pem_c = utils.pem.ApprisePEMController(path=str(tmpdir0), asset=asset)
    assert pem_c.public_keyfile('test1', 'test2') == pub_keyfile
    assert pem_c.private_keyfile('test1', 'test2') == prv_keyfile

    pem_c = utils.pem.ApprisePEMController(
        path=str(tmpdir0), name='pub1', asset=asset)
    assert pem_c.public_key(autogen=True)

    pem_c = utils.pem.ApprisePEMController(
        path=str(tmpdir0), name='pub2', asset=asset)
    assert pem_c.private_key(autogen=True)

    #
    # Auto key generation turned on
    #
    asset = AppriseAsset(
        storage_mode=PersistentStoreMode.MEMORY,
        storage_path=str(tmpdir0),
        pem_autogen=True,
    )
    pem_c = utils.pem.ApprisePEMController(path=str(tmpdir0), asset=asset)
    assert pem_c.load_public_key(path=pub_keyfile) is True
    pem_c = utils.pem.ApprisePEMController(path=None, asset=asset)
    assert pem_c.load_public_key(path=pub_keyfile) is True

    tmpdir1 = tmpdir.mkdir('tmp01')

    # Currently no files here
    assert os.listdir(str(tmpdir1)) == []

    asset = AppriseAsset(
        storage_mode=PersistentStoreMode.MEMORY,
        storage_path=str(tmpdir1),
        pem_autogen=False,
    )

    # Auto-Gen is turned off, so weare not successful here
    pem_c = utils.pem.ApprisePEMController(path=None, asset=asset)
    assert pem_c.public_key() is None
    assert pem_c.private_key() is None
    pem_c = utils.pem.ApprisePEMController(path=str(tmpdir1), asset=asset)
    assert pem_c.public_key() is None
    assert pem_c.private_key() is None

    asset = AppriseAsset(
        storage_mode=PersistentStoreMode.FLUSH,
        storage_path=str(tmpdir1),
        pem_autogen=True,
    )
    pem_c = utils.pem.ApprisePEMController(path=str(tmpdir1), asset=asset)
    # Generate ourselves a private key
    assert pem_c.public_key() is not None
    assert pem_c.private_key() is not None
    pub_keyfile = os.path.join(str(tmpdir1), 'public_key.pem')
    prv_keyfile = os.path.join(str(tmpdir1), 'private_key.pem')
    assert os.path.isfile(pub_keyfile)
    assert os.path.isfile(prv_keyfile)

    with open(pub_keyfile, 'w') as f:
        f.write('garbage')

    pem_c = utils.pem.ApprisePEMController(path=str(tmpdir1), asset=asset)
    # we can still load our data as the public key is generated
    # from the private
    assert pem_c.public_key() is not None
    assert pem_c.private_key() is not None

    tmpdir2 = tmpdir.mkdir('tmp02')
    pem_c = utils.pem.ApprisePEMController(path=str(tmpdir2), asset=asset)
    pub_keyfile = os.path.join(str(tmpdir2), 'public_key.pem')
    prv_keyfile = os.path.join(str(tmpdir2), 'private_key.pem')
    assert not os.path.isfile(pub_keyfile)
    assert not os.path.isfile(prv_keyfile)

    #
    # Public Key Edge Case Tests
    #
    with \
        mock.patch.object(
            pem_c, 'public_keyfile', side_effect=[None, pub_keyfile]) \
        as mock_keyfile, \
        mock.patch.object(
            pem_c, 'keygen', side_effect=lambda *_, **__: setattr(
                pem_c, '_ApprisePEMController__public_key',
                pubkey_ref) or True) as mock_keygen, \
        mock.patch.object(
            pem_c, 'load_public_key', return_value=True):

        result = pem_c.public_key()
        assert result is pubkey_ref
        assert mock_keyfile.call_count == 2
        mock_keygen.assert_called_once()

    # - First call: None → triggers keygen
    # - Second call (recursive): None → causes fallback
    public_keyfile_side_effect = [None, None]

    with mock.patch.object(
            pem_c, 'public_keyfile', side_effect=public_keyfile_side_effect) \
         as mock_keyfile, \
         mock.patch.object(pem_c, 'keygen', return_value=True) \
         as mock_keygen, \
         mock.patch.object(pem_c, 'load_public_key', return_value=False) \
         as mock_load:

        # Ensure no key is preset initially
        setattr(pem_c, '_ApprisePEMController__public_key', None)

        result = pem_c.public_key()
        assert result is None
        # Once in outer call, once in recursive
        assert mock_keyfile.call_count == 2
        mock_keygen.assert_called_once()
        mock_load.assert_not_called()

    #
    # Private Key Edge Case Tests
    #
    with \
        mock.patch.object(
            pem_c, 'private_keyfile', side_effect=[None, prv_keyfile]) \
        as mock_keyfile, \
        mock.patch.object(
            pem_c, 'keygen', side_effect=lambda *_, **__: setattr(
                pem_c, '_ApprisePEMController__private_key',
                prvkey_ref) or True) as mock_keygen, \
        mock.patch.object(
            pem_c, 'load_private_key', return_value=True):

        result = pem_c.private_key()
        assert result is prvkey_ref
        assert mock_keyfile.call_count == 2
        mock_keygen.assert_called_once()

    # - First call: None → triggers keygen
    # - Second call (recursive): None → causes fallback
    private_keyfile_side_effect = [None, None]

    with mock.patch.object(
            pem_c, 'private_keyfile',
            side_effect=private_keyfile_side_effect) as mock_keyfile, \
         mock.patch.object(pem_c, 'keygen', return_value=True) \
         as mock_keygen, \
         mock.patch.object(pem_c, 'load_private_key', return_value=False) \
         as mock_load:

        # Ensure no key is preset initially
        setattr(pem_c, '_ApprisePEMController__private_key', None)

        result = pem_c.private_key()
        assert result is None
        # Once in outer call, once in recursive
        assert mock_keyfile.call_count == 2
        mock_keygen.assert_called_once()
        mock_load.assert_not_called()


@pytest.mark.skipif(
    'cryptography' in sys.modules,
    reason="Requires that cryptography NOT be installed")
def test_utils_pem_general_without_c(tmpdir):
    """
    Utils:PEM Without cryptography

    """

    tmpdir0 = tmpdir.mkdir('tmp00')

    # Currently no files here
    assert os.listdir(str(tmpdir0)) == []

    asset = AppriseAsset(
        storage_mode=PersistentStoreMode.MEMORY,
        storage_path=str(tmpdir0),
        pem_autogen=False,
    )

    # Create a PEM Controller
    pem_c = utils.pem.ApprisePEMController(path=None, asset=asset)

    # cryptography library missing poses issues with library useage
    with pytest.raises(utils.pem.ApprisePEMException):
        pem_c.public_keyfile()

    with pytest.raises(utils.pem.ApprisePEMException):
        pem_c.public_key()

    with pytest.raises(utils.pem.ApprisePEMException):
        pem_c.x962_str

    with pytest.raises(utils.pem.ApprisePEMException):
        pem_c.encrypt("message")

    with pytest.raises(utils.pem.ApprisePEMException):
        pem_c.keygen()

    asset = AppriseAsset(
        storage_mode=PersistentStoreMode.FLUSH,
        storage_path=str(tmpdir0),
        pem_autogen=False,
    )

    # No new files
    assert os.listdir(str(tmpdir0)) == []

    # Our asset is now write mode, so we will be able to generate a key
    pem_c = utils.pem.ApprisePEMController(path=str(tmpdir0), asset=asset)
    # Nothing to lookup
    with pytest.raises(utils.pem.ApprisePEMException):
        pem_c.private_keyfile()

    with pytest.raises(utils.pem.ApprisePEMException):
        pem_c.private_key()

    with pytest.raises(utils.pem.ApprisePEMException):
        pem_c.x962_str

    with pytest.raises(utils.pem.ApprisePEMException):
        pem_c.encrypt("message")

    # Keys can not be generated in memory mode
    with pytest.raises(utils.pem.ApprisePEMException):
        pem_c.keygen()

    # No files loaded
    assert os.listdir(str(tmpdir0)) == []

    with pytest.raises(utils.pem.ApprisePEMException):
        pem_c.public_keyfile()

    with pytest.raises(utils.pem.ApprisePEMException):
        pem_c.public_key()

    with pytest.raises(utils.pem.ApprisePEMException):
        pem_c.x962_str

    with pytest.raises(utils.pem.ApprisePEMException):
        pem_c.encrypt("message")

    with pytest.raises(utils.pem.ApprisePEMException):
        pem_c.decrypt("abcd==")