Russell Ballestrini
1847a2fee6
Tests for: - async_web_fetcher: URL handling, media extraction, crawl modes, scoring - domain_vault: VaultManager, HTML/Media/Linkpeek vaults - screenshot: ScreenshotCapture configuration and availability - storage: ImageVault with MD5 deduplication 185 tests total, all passing.
364 lines
11 KiB
Python
364 lines
11 KiB
Python
"""
|
|
Tests for domain_vault module.
|
|
|
|
Tests domain-based vault system for HTML, media, and screenshots.
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import Mock, AsyncMock, patch, MagicMock
|
|
import tempfile
|
|
import shutil
|
|
import os
|
|
import hashlib
|
|
from pathlib import Path
|
|
|
|
from domain_vault import (
|
|
GitRepo,
|
|
DomainHtmlVault,
|
|
DomainMediaVault,
|
|
DomainLinkpeekVault,
|
|
VaultManager,
|
|
)
|
|
|
|
|
|
class TestGitRepo:
|
|
"""Test GitRepo class."""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.repo_path = Path(self.temp_dir) / "test_repo"
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_git_repo_initialization(self):
|
|
"""Test GitRepo initialization."""
|
|
repo = GitRepo(self.repo_path)
|
|
|
|
assert repo.path == self.repo_path
|
|
assert repo.use_lfs is False
|
|
|
|
def test_git_repo_with_lfs(self):
|
|
"""Test GitRepo with LFS enabled."""
|
|
repo = GitRepo(self.repo_path, use_lfs=True)
|
|
|
|
assert repo.use_lfs is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_git_repo_init_creates_directory(self):
|
|
"""Test that init creates the directory."""
|
|
repo = GitRepo(self.repo_path)
|
|
|
|
with patch.object(repo, '_run', new_callable=AsyncMock) as mock_run:
|
|
mock_run.return_value = (b'', b'')
|
|
await repo.init()
|
|
|
|
assert self.repo_path.exists()
|
|
|
|
|
|
class TestDomainHtmlVault:
|
|
"""Test DomainHtmlVault class."""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.vault_base = os.path.join(self.temp_dir, "vault")
|
|
self.domain = "example.com"
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_html_vault_initialization(self):
|
|
"""Test DomainHtmlVault initialization."""
|
|
vault = DomainHtmlVault(self.vault_base, self.domain)
|
|
|
|
assert vault.domain == self.domain.lower()
|
|
assert vault.media_base_url == '/media'
|
|
|
|
def test_html_vault_media_base_url(self):
|
|
"""Test custom media base URL."""
|
|
vault = DomainHtmlVault(
|
|
self.vault_base,
|
|
self.domain,
|
|
media_base_url="/assets"
|
|
)
|
|
|
|
assert vault.media_base_url == "/assets"
|
|
|
|
def test_get_media_vault_url(self):
|
|
"""Test media URL rewriting."""
|
|
vault = DomainHtmlVault(self.vault_base, self.domain)
|
|
|
|
media_url = "https://example.com/images/photo.jpg"
|
|
result = vault.get_media_vault_url(media_url)
|
|
|
|
assert vault.media_base_url in result
|
|
assert "photo.jpg" in result or "images" in result
|
|
|
|
def test_rewrite_media_urls(self):
|
|
"""Test HTML media URL rewriting."""
|
|
vault = DomainHtmlVault(self.vault_base, self.domain)
|
|
|
|
html = '<img src="https://example.com/photo.jpg">'
|
|
mappings = {"https://example.com/photo.jpg": "/media/abc123.jpg"}
|
|
|
|
result = vault.rewrite_media_urls(html, "https://example.com/", mappings)
|
|
|
|
# Should contain the rewritten URL
|
|
assert "/media/abc123.jpg" in result
|
|
|
|
|
|
class TestDomainMediaVault:
|
|
"""Test DomainMediaVault class."""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.vault_base = os.path.join(self.temp_dir, "vault")
|
|
self.domain = "example.com"
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_media_vault_initialization(self):
|
|
"""Test DomainMediaVault initialization."""
|
|
vault = DomainMediaVault(self.vault_base, self.domain)
|
|
|
|
assert vault.domain == self.domain.lower()
|
|
|
|
|
|
class TestDomainLinkpeekVault:
|
|
"""Test DomainLinkpeekVault class."""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.vault_base = os.path.join(self.temp_dir, "vault")
|
|
self.domain = "example.com"
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_linkpeek_vault_initialization(self):
|
|
"""Test DomainLinkpeekVault initialization."""
|
|
vault = DomainLinkpeekVault(self.vault_base, self.domain)
|
|
|
|
assert vault.domain == self.domain.lower()
|
|
|
|
def test_url_to_screenshot_path(self):
|
|
"""Test URL to screenshot path conversion."""
|
|
vault = DomainLinkpeekVault(self.vault_base, self.domain)
|
|
|
|
url = "https://example.com/page"
|
|
path = vault.url_to_screenshot_path(url)
|
|
|
|
# Should generate a path ending in .png
|
|
assert path.endswith('.png')
|
|
|
|
|
|
class TestVaultManager:
|
|
"""Test VaultManager class."""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.html_base = os.path.join(self.temp_dir, "html_vault")
|
|
self.media_base = os.path.join(self.temp_dir, "media_vault")
|
|
self.linkpeek_base = os.path.join(self.temp_dir, "linkpeek_vault")
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_vault_manager_initialization(self):
|
|
"""Test VaultManager initialization."""
|
|
manager = VaultManager(
|
|
html_vault_base=self.html_base,
|
|
media_vault_base=self.media_base,
|
|
linkpeek_vault_base=self.linkpeek_base,
|
|
)
|
|
|
|
assert str(manager.html_base) == self.html_base
|
|
|
|
def test_vault_manager_default_init(self):
|
|
"""Test VaultManager with defaults."""
|
|
manager = VaultManager()
|
|
|
|
assert manager.html_base == Path('html_vault')
|
|
assert manager.media_base == Path('media_vault')
|
|
assert manager.linkpeek_base == Path('linkpeek_vault')
|
|
|
|
def test_get_html_vault(self):
|
|
"""Test getting HTML vault for domain."""
|
|
manager = VaultManager(
|
|
html_vault_base=self.html_base,
|
|
media_vault_base=self.media_base,
|
|
linkpeek_vault_base=self.linkpeek_base,
|
|
)
|
|
|
|
vault = manager.get_html_vault("example.com")
|
|
|
|
assert isinstance(vault, DomainHtmlVault)
|
|
assert vault.domain == "example.com"
|
|
|
|
def test_get_html_vault_caches(self):
|
|
"""Test that HTML vaults are cached."""
|
|
manager = VaultManager(
|
|
html_vault_base=self.html_base,
|
|
media_vault_base=self.media_base,
|
|
linkpeek_vault_base=self.linkpeek_base,
|
|
)
|
|
|
|
vault1 = manager.get_html_vault("example.com")
|
|
vault2 = manager.get_html_vault("example.com")
|
|
|
|
assert vault1 is vault2
|
|
|
|
def test_get_html_vault_case_insensitive(self):
|
|
"""Test that domain lookup is case insensitive."""
|
|
manager = VaultManager(
|
|
html_vault_base=self.html_base,
|
|
media_vault_base=self.media_base,
|
|
linkpeek_vault_base=self.linkpeek_base,
|
|
)
|
|
|
|
vault1 = manager.get_html_vault("Example.COM")
|
|
vault2 = manager.get_html_vault("example.com")
|
|
|
|
assert vault1 is vault2
|
|
assert vault1.domain == "example.com"
|
|
|
|
def test_get_media_vault(self):
|
|
"""Test getting media vault for domain."""
|
|
manager = VaultManager(
|
|
html_vault_base=self.html_base,
|
|
media_vault_base=self.media_base,
|
|
linkpeek_vault_base=self.linkpeek_base,
|
|
)
|
|
|
|
vault = manager.get_media_vault("example.com")
|
|
|
|
assert isinstance(vault, DomainMediaVault)
|
|
assert vault.domain == "example.com"
|
|
|
|
def test_get_linkpeek_vault(self):
|
|
"""Test getting linkpeek vault for domain."""
|
|
manager = VaultManager(
|
|
html_vault_base=self.html_base,
|
|
media_vault_base=self.media_base,
|
|
linkpeek_vault_base=self.linkpeek_base,
|
|
)
|
|
|
|
vault = manager.get_linkpeek_vault("example.com")
|
|
|
|
assert isinstance(vault, DomainLinkpeekVault)
|
|
assert vault.domain == "example.com"
|
|
|
|
def test_get_vaults_for_url(self):
|
|
"""Test getting all vaults for a URL."""
|
|
manager = VaultManager(
|
|
html_vault_base=self.html_base,
|
|
media_vault_base=self.media_base,
|
|
linkpeek_vault_base=self.linkpeek_base,
|
|
)
|
|
|
|
html_vault, media_vault, linkpeek_vault = manager.get_vaults_for_url(
|
|
"https://example.com/page"
|
|
)
|
|
|
|
assert isinstance(html_vault, DomainHtmlVault)
|
|
assert isinstance(media_vault, DomainMediaVault)
|
|
assert isinstance(linkpeek_vault, DomainLinkpeekVault)
|
|
assert html_vault.domain == "example.com"
|
|
|
|
def test_get_linkpeek_url(self):
|
|
"""Test generating linkpeek URL."""
|
|
manager = VaultManager(
|
|
html_vault_base=self.html_base,
|
|
media_vault_base=self.media_base,
|
|
linkpeek_vault_base=self.linkpeek_base,
|
|
linkpeek_base_url="/screenshots"
|
|
)
|
|
|
|
url = manager.get_linkpeek_url("https://example.com/page")
|
|
|
|
assert "/screenshots/" in url
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_domains_empty(self):
|
|
"""Test listing domains on empty vault."""
|
|
manager = VaultManager(
|
|
html_vault_base=self.html_base,
|
|
media_vault_base=self.media_base,
|
|
linkpeek_vault_base=self.linkpeek_base,
|
|
)
|
|
|
|
# Create the base directory
|
|
os.makedirs(self.html_base, exist_ok=True)
|
|
|
|
domains = await manager.list_domains("html")
|
|
|
|
assert domains == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_all_stats(self):
|
|
"""Test getting all stats."""
|
|
manager = VaultManager(
|
|
html_vault_base=self.html_base,
|
|
media_vault_base=self.media_base,
|
|
linkpeek_vault_base=self.linkpeek_base,
|
|
)
|
|
|
|
# Create base directories
|
|
os.makedirs(self.html_base, exist_ok=True)
|
|
os.makedirs(self.media_base, exist_ok=True)
|
|
os.makedirs(self.linkpeek_base, exist_ok=True)
|
|
|
|
stats = await manager.get_all_stats()
|
|
|
|
assert 'html_vaults' in stats
|
|
assert 'media_vaults' in stats
|
|
assert 'linkpeek_vaults' in stats
|
|
|
|
|
|
class TestVaultManagerDomainExtraction:
|
|
"""Test domain extraction from URLs."""
|
|
|
|
def setup_method(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.manager = VaultManager(
|
|
html_vault_base=os.path.join(self.temp_dir, "html"),
|
|
media_vault_base=os.path.join(self.temp_dir, "media"),
|
|
linkpeek_vault_base=os.path.join(self.temp_dir, "linkpeek"),
|
|
)
|
|
|
|
def teardown_method(self):
|
|
if os.path.exists(self.temp_dir):
|
|
shutil.rmtree(self.temp_dir)
|
|
|
|
def test_extract_domain_basic(self):
|
|
"""Test extracting domain from basic URL."""
|
|
vault = self.manager.get_html_vault("example.com")
|
|
assert vault.domain == "example.com"
|
|
|
|
def test_vaults_for_subdomain(self):
|
|
"""Test getting vaults for subdomain."""
|
|
html, media, linkpeek = self.manager.get_vaults_for_url(
|
|
"https://sub.example.com/page"
|
|
)
|
|
|
|
assert html.domain == "sub.example.com"
|
|
assert media.domain == "sub.example.com"
|
|
|
|
def test_vaults_for_url_with_port(self):
|
|
"""Test getting vaults for URL with port."""
|
|
html, media, linkpeek = self.manager.get_vaults_for_url(
|
|
"https://example.com:8080/page"
|
|
)
|
|
|
|
# Domain includes port
|
|
assert "example.com" in html.domain
|
|
|
|
|
|
if __name__ == '__main__':
|
|
pytest.main([__file__, '-v'])
|