import os import pytest from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import rsa from monday_com_orca_backend.connectors.snowflake_db import constants from monday_com_orca_backend.connectors.snowflake_db.utils import ( get_private_key_content, ) class TestGetPrivateKeyContent: @pytest.fixture(scope="class") def pem_key(self): private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) pem = private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption(), ) return pem.decode("utf-8") @pytest.fixture(scope="class") def base64_key(self, pem_key): return "".join( line for line in pem_key.splitlines() if not line.startswith("-----") ) def assert_wrappped(self, result): assert isinstance(result, bytes) assert result.startswith( constants.PRIVATE_KEY_HEADER ), "Private key content should start with the correct header." assert result.endswith( constants.PRIVATE_KEY_FOOTER ), "Private key content should end with the correct footer." assert not result.endswith( b"\n", 0, -1 ), "Private key content should not end with a newline character." @pytest.mark.parametrize("trailing", ["", "\n", " ", "\n "]) def test_pem_input_variants(self, pem_key, trailing): key = pem_key + trailing result = get_private_key_content(key) self.assert_wrappped(result) def test_file_path_reads_file(self, tmp_path, pem_key): file_path = tmp_path / "key.pem" file_path.write_text(pem_key) result = get_private_key_content(str(file_path)) self.assert_wrappped(result) def test_base64_key_reconstructs_pem(self, base64_key): result = get_private_key_content(base64_key) self.assert_wrappped(result) assert base64_key[:10].encode() in result @pytest.mark.parametrize("pad", ["", " ", "\n", "\n "]) def test_strip_whitespace(self, pem_key, pad): key = f"{pad}{pem_key}{pad}" result = get_private_key_content(key) self.assert_wrappped(result) def test_nonexistent_file_treated_as_base64(self, monkeypatch, base64_key): monkeypatch.setattr(os.path, "exists", lambda x: False) result = get_private_key_content(base64_key) self.assert_wrappped(result) def test_empty_string_raises(self): with pytest.raises(ValueError, match="Private key cannot be empty."): get_private_key_content("") def test_invalid_path_raises(self, monkeypatch): monkeypatch.setattr(os.path, "exists", lambda x: True) import builtins monkeypatch.setattr( builtins, "open", lambda *a, **kw: (_ for _ in ()).throw(FileNotFoundError) ) with pytest.raises(FileNotFoundError): get_private_key_content("some_nonexistent_file.pem") @pytest.mark.parametrize("input_type", ["pem", "base64"]) def test_roundtrip_consistency(self, pem_key, base64_key, input_type): if input_type == "pem": result = get_private_key_content(pem_key) else: result = get_private_key_content(base64_key) self.assert_wrappped(result)