leocrm/tests/test_marketplace.py

"""Tests for marketplace plugin system: signature, quarantine, allowlist."""

from __future__ import annotations

import hashlib
import tempfile
from pathlib import Path

import pytest

from app.plugins.signature import PluginSignature
from app.plugins.quarantine import (
    QuarantineError,
    _check_dangerous_imports,
    _check_migration_sql,
    _validate_manifest,
)


class TestPluginSignature:
    def test_compute_hash(self, tmp_path):
        """compute_hash returns a valid SHA-256 hex string."""
        test_file = tmp_path / "test.zip"
        test_file.write_bytes(b"test content")
        h = PluginSignature.compute_hash(test_file)
        assert len(h) == 64  # SHA-256 hex
        assert h == hashlib.sha256(b"test content").hexdigest()

    def test_verify_signature_without_pynacl(self, tmp_path):
        """verify_signature returns False if PyNaCl is not installed."""
        test_file = tmp_path / "test.zip"
        test_file.write_bytes(b"test")
        # Without PyNaCl installed, returns False
        result = PluginSignature.verify_signature(test_file, b"sig", b"key")
        assert result in (False, True)  # Depends on whether pynacl is installed


class TestQuarantineValidation:
    def test_validate_manifest_valid(self, tmp_path):
        """_validate_manifest passes for a valid plugin structure."""
        plugin_dir = tmp_path / "test_plugin"
        plugin_dir.mkdir()
        (plugin_dir / "plugin.py").write_text(
            "from app.plugins.base import BasePlugin\n"
            "from app.plugins.manifest import PluginManifest\n"
            "class TestPlugin(BasePlugin):\n"
            "    manifest = PluginManifest(name='test', version='1.0.0', display_name='Test')\n"
        )
        result = _validate_manifest(plugin_dir)
        assert result["has_manifest"] is True

    def test_validate_manifest_missing(self, tmp_path):
        """_validate_manifest raises for missing plugin.py."""
        with pytest.raises(QuarantineError, match="plugin.py or __init__.py"):
            _validate_manifest(tmp_path)

    def test_validate_manifest_no_manifest(self, tmp_path):
        """_validate_manifest raises when PluginManifest is missing."""
        plugin_dir = tmp_path / "test_plugin"
        plugin_dir.mkdir()
        (plugin_dir / "plugin.py").write_text("print('hello')")
        with pytest.raises(QuarantineError, match="PluginManifest"):
            _validate_manifest(plugin_dir)

    def test_check_dangerous_imports_clean(self, tmp_path):
        """_check_dangerous_imports returns empty for safe code."""
        plugin_dir = tmp_path / "safe_plugin"
        plugin_dir.mkdir()
        (plugin_dir / "plugin.py").write_text(
            "import logging\n"
            "from app.plugins.base import BasePlugin\n"
        )
        result = _check_dangerous_imports(plugin_dir)
        assert result == []

    def test_check_dangerous_imports_found(self, tmp_path):
        """_check_dangerous_imports detects dangerous patterns."""
        plugin_dir = tmp_path / "dangerous_plugin"
        plugin_dir.mkdir()
        (plugin_dir / "plugin.py").write_text(
            "import os\n"
            "os.system('rm -rf /')\n"
        )
        result = _check_dangerous_imports(plugin_dir)
        assert len(result) > 0
        assert any("os.system" in r for r in result)

    def test_check_migration_sql_no_migrations(self, tmp_path):
        """_check_migration_sql returns empty when no migrations dir."""
        result = _check_migration_sql(tmp_path)
        assert result == []

    def test_check_migration_sql_valid(self, tmp_path):
        """_check_migration_sql passes for valid SQL with tenant_id."""
        migrations = tmp_path / "migrations"
        migrations.mkdir()
        (migrations / "0001_initial.sql").write_text(
            "CREATE TABLE items (id UUID, tenant_id UUID NOT NULL);\n"
        )
        result = _check_migration_sql(tmp_path)
        assert result == []

    def test_check_migration_sql_missing_tenant_id(self, tmp_path):
        """_check_migration_sql detects missing tenant_id."""
        migrations = tmp_path / "migrations"
        migrations.mkdir()
        (migrations / "0001_initial.sql").write_text(
            "CREATE TABLE items (id UUID);\n"
        )
        result = _check_migration_sql(tmp_path)
        assert len(result) > 0
        assert "tenant_id" in result[0]

    def test_check_migration_sql_drop_database(self, tmp_path):
        """_check_migration_sql detects DROP DATABASE."""
        migrations = tmp_path / "migrations"
        migrations.mkdir()
        (migrations / "0001_initial.sql").write_text(
            "DROP DATABASE leocrm;\n"
        )
        result = _check_migration_sql(tmp_path)
        assert len(result) > 0
        assert "DROP" in result[0]


class TestManifestMarketplaceFields:
    def test_manifest_has_marketplace_fields(self):
        """PluginManifest has marketplace fields."""
        from app.plugins.manifest import PluginManifest
        m = PluginManifest(
            name="test",
            version="1.0.0",
            display_name="Test",
            author="Test Author",
            license="MIT",
            min_app_version="1.0.0",
        )
        assert m.author == "Test Author"
        assert m.license == "MIT"
        assert m.min_app_version == "1.0.0"
        assert m.contract_version == "1.0.0"
        assert m.hooks == []

    def test_manifest_marketplace_optional_fields(self):
        """PluginManifest marketplace fields have defaults."""
        from app.plugins.manifest import PluginManifest
        m = PluginManifest(
            name="test",
            version="1.0.0",
            display_name="Test",
        )
        assert m.author == ""
        assert m.homepage == ""
        assert m.license == "MIT"
        assert m.price == 0.0
        assert m.screenshots == []
        assert m.marketplace_tags == []