Select a provider, root, and item to see details.
-
-
-
+
+
+
Current Location:
+
+ No folder selected
+
+
+
+
+
+ Select a folder to enable scanning. All scans run as background tasks with progress tracking.
+
@@ -236,8 +237,8 @@
Start Background Scan
const provList = document.getElementById('prov-list');
const provCrumb = document.getElementById('prov-crumb');
const provPanel = document.getElementById('prov-panel-content');
- const provScanForm = document.getElementById('prov-scan-form');
- const provScanMsg = document.getElementById('prov-scan-msg');
+ const provScanBtn = document.getElementById('prov-scan-btn');
+ const provCurrentPath = document.getElementById('prov-current-path');
const btnROC = document.getElementById('open-rocrate');
const btnROCClose = document.getElementById('close-rocrate');
const rocFrame = document.getElementById('rocrate-frame');
@@ -342,7 +343,18 @@
Start Background Scan
}).join('');
provList.innerHTML = rows || '
| Empty folder. |
';
attachProvHandlers();
- } catch(e){ provList.innerHTML = '
| Browse failed. |
'; }
+
+ // Update current path display and enable scan button
+ if (provCurrentPath) {
+ provCurrentPath.textContent = fullPath || '(root)';
+ }
+ if (provScanBtn) {
+ provScanBtn.disabled = false;
+ }
+ } catch(e){
+ provList.innerHTML = '
| Browse failed. |
';
+ if (provScanBtn) provScanBtn.disabled = true;
+ }
}
function attachProvHandlers(){
@@ -433,75 +445,65 @@
Start Background Scan
if (chkFast) chkFast.addEventListener('change', reBrowse);
if (inpDepth) inpDepth.addEventListener('change', reBrowse);
}
- if (provScanForm){
- provScanForm.addEventListener('submit', async (e) => {
- e.preventDefault();
- // Derive values from UI at submit-time to avoid race with async init
+
+ // Unified scan button handler - uses background tasks
+ if (provScanBtn){
+ provScanBtn.addEventListener('click', async () => {
+ // Get current browse context
const provId = (provSelect && provSelect.value) || currentProv || 'local_fs';
const rootId = (rootSelect && rootSelect.value) || currentRoot || '/';
const inputPath = (provPathInput && provPathInput.value && provPathInput.value.trim()) || '';
const relOrAbs = inputPath || currentPath || '';
- // Compose full scan target for Rclone; keep other providers untouched
+
+ // Compose full scan path
let scanPath = relOrAbs || rootId || '/';
if (provId === 'rclone') {
- scanPath = composePath(provId, rootId, relOrAbs); // e.g., "dropbox:AIPT"
+ scanPath = composePath(provId, rootId, relOrAbs);
}
- currentProv = provId;
- currentRoot = rootId;
- currentPath = relOrAbs;
- if (!provId || !scanPath){ provScanMsg.textContent = 'Select a provider and folder first.'; return; }
- const recursive = document.getElementById('prov-scan-recursive').checked;
- const chkFast = document.getElementById('prov-browse-fast-list');
- const fastList = !!(chkFast && chkFast.checked);
- const btn = provScanForm.querySelector('button[type="submit"]');
- if (btn) { btn.disabled = true; btn.textContent = 'Scanningβ¦'; }
- provScanMsg.textContent = `Starting scan for ${scanPath}β¦`;
- // Add a local pseudo-task so status appears with other progress bars
- const localId = 'provscan-' + Date.now();
- const localTask = { id: localId, type: 'scan', status: 'running', path: scanPath, processed: 0, total: null, progress: 0 };
- try { (window.scidkLocalTasks||[]).push(localTask); } catch(_) { /* ignore */ }
- fetchTasks(); // trigger re-render with local task
+
+ if (!provId || !scanPath){
+ alert('Please select a provider and folder first.');
+ return;
+ }
+
+ // Populate background scan form and trigger it
+ const bgPathInput = document.getElementById('scan-path');
+ const bgRecursive = document.getElementById('scan-recursive');
+ if (bgPathInput) bgPathInput.value = scanPath;
+ if (bgRecursive) bgRecursive.checked = true;
+
+ // Trigger background scan via tasks API (same as "Start Background Scan" button)
+ const recursive = true;
+ provScanBtn.disabled = true;
+ provScanBtn.textContent = 'Starting scan...';
+
try {
- const r = await fetch('/api/scan', { method: 'POST', headers: { 'Content-Type':'application/json' }, body: JSON.stringify({ provider_id: provId, root_id: rootId||'/', path: scanPath, recursive, fast_list: fastList }) });
- const ctype = (r.headers && r.headers.get('content-type')) || '';
- let j = null;
- if (ctype.includes('application/json')){
- try { j = await r.json(); } catch(_) { j = null; }
- } else {
- // Non-JSON response (proxy/gateway error). Read text for clarity
- try { const txt = await r.text(); throw new Error(`HTTP ${r.status}: ${txt}`); } catch(e){ throw e; }
- }
- if (r.ok && j){
- const files = j.scanned || 0;
- const folders = (j.folder_count !== undefined) ? j.folder_count : undefined;
- const dur = j.duration_sec ? (Math.round(j.duration_sec*10)/10+'s') : '';
- let msg = `Scan complete: ${j.scan_id} β files: ${files}`;
- if (folders !== undefined) msg += `, folders: ${folders}`;
- if (files === 0 && folders > 0 && !recursive){ msg += ' β Only folders found. Enable Recursive to include files in subfolders.'; }
- if (dur) msg += ` (${dur})`;
- provScanMsg.textContent = msg;
- // Mark local task completed
- localTask.status = 'completed';
- localTask.processed = files;
- localTask.total = files || localTask.processed;
- localTask.progress = 1;
- localTask.scan_id = j.scan_id;
+ const payload = {
+ type: 'scan',
+ path: scanPath,
+ recursive,
+ provider_id: provId,
+ root_id: rootId
+ };
+ const r = await fetch('/api/tasks', {
+ method: 'POST',
+ headers: { 'Content-Type':'application/json' },
+ body: JSON.stringify(payload)
+ });
+
+ if (r.status === 202){
+ startPolling();
+ fetchTasks();
+ alert(`Background scan started for: ${scanPath}\nCheck progress in "Scans Summary" section below.`);
} else {
- const err = (j && (j.error||j.message)) || `HTTP ${r.status}`;
- provScanMsg.textContent = `Scan error: ${err}`;
- localTask.status = 'error';
- localTask.error = err;
- localTask.progress = 1;
+ const j = await r.json();
+ alert('Scan error: ' + (j.error || r.status));
}
} catch(err){
- provScanMsg.textContent = 'Scan error: ' + err;
- localTask.status = 'error';
- localTask.error = (err && err.message) ? err.message : String(err);
- localTask.progress = 1;
- }
- finally {
- if (btn) { btn.disabled = false; btn.textContent = 'Scan'; }
- fetchTasks(); // refresh tasks view to reflect final status
+ alert('Scan error: ' + err);
+ } finally {
+ provScanBtn.disabled = false;
+ provScanBtn.textContent = 'π Scan This Folder';
}
});
}
diff --git a/tests/test_files_page_e2e.py b/tests/test_files_page_e2e.py
new file mode 100644
index 0000000..6b87ae4
--- /dev/null
+++ b/tests/test_files_page_e2e.py
@@ -0,0 +1,308 @@
+"""
+End-to-end tests for Files page UX workflows.
+
+Validates the consolidated scan functionality and browser-to-scan integration.
+"""
+import os
+import time
+from pathlib import Path
+import pytest
+from bs4 import BeautifulSoup
+
+
+def test_files_page_loads_successfully():
+ """Test that the Files page loads without errors."""
+ from scidk.app import create_app
+ app = create_app()
+ app.config['TESTING'] = True
+
+ with app.test_client() as client:
+ resp = client.get('/datasets')
+ assert resp.status_code == 200
+ assert b'Files' in resp.data
+ assert b'Provider' in resp.data
+
+
+def test_scan_button_uses_background_tasks_only():
+ """Verify that the scan button uses /api/tasks, not /api/scan."""
+ from scidk.app import create_app
+ app = create_app()
+ app.config['TESTING'] = True
+
+ with app.test_client() as client:
+ resp = client.get('/datasets')
+ assert resp.status_code == 200
+
+ # Check that the template has the new unified scan button
+ html = resp.data.decode('utf-8')
+ assert 'prov-scan-btn' in html
+ assert 'π Scan This Folder' in html
+
+ # Check that the old sync scan form is removed
+ assert 'prov-scan-form' not in html
+ assert 'prov-scan-recursive' not in html # old checkbox removed
+
+
+def test_browse_and_scan_integration(tmp_path: Path):
+ """Test the full workflow: browse folder β scan it via background task."""
+ from scidk.app import create_app
+ app = create_app()
+ app.config['TESTING'] = True
+
+ # Create test directory
+ test_dir = tmp_path / 'test_project'
+ test_dir.mkdir()
+ (test_dir / 'file1.txt').write_text('content1', encoding='utf-8')
+ (test_dir / 'file2.txt').write_text('content2', encoding='utf-8')
+ (test_dir / 'subdir').mkdir()
+ (test_dir / 'subdir' / 'file3.txt').write_text('content3', encoding='utf-8')
+
+ with app.test_client() as client:
+ # Browse the directory
+ browse_resp = client.get(f'/api/browse?provider_id=local_fs&root_id=/&path={str(test_dir)}')
+ assert browse_resp.status_code == 200
+ browse_data = browse_resp.get_json()
+ assert 'entries' in browse_data
+ assert len(browse_data['entries']) >= 3 # 2 files + 1 subdir
+
+ # Trigger scan via background task (unified mechanism)
+ scan_resp = client.post('/api/tasks', json={
+ 'type': 'scan',
+ 'path': str(test_dir),
+ 'recursive': True,
+ 'provider_id': 'local_fs',
+ 'root_id': '/'
+ })
+ assert scan_resp.status_code == 202 # Accepted
+ scan_data = scan_resp.get_json()
+ assert 'task_id' in scan_data
+ task_id = scan_data['task_id']
+
+ # Poll for task completion (max 10 seconds)
+ max_wait = 10
+ start_time = time.time()
+ task_completed = False
+
+ while time.time() - start_time < max_wait:
+ task_resp = client.get(f'/api/tasks/{task_id}')
+ assert task_resp.status_code == 200
+ task_data = task_resp.get_json()
+
+ if task_data['status'] == 'completed':
+ task_completed = True
+ assert task_data['processed'] >= 3
+ break
+
+ time.sleep(0.5)
+
+ assert task_completed, "Scan task did not complete in time"
+
+
+def test_scan_history_unified_display(tmp_path: Path):
+ """Test that all scans appear in unified history."""
+ from scidk.app import create_app
+ app = create_app()
+ app.config['TESTING'] = True
+
+ test_dir = tmp_path / 'scan_test'
+ test_dir.mkdir()
+ (test_dir / 'test.txt').write_text('test', encoding='utf-8')
+
+ with app.test_client() as client:
+ # Create first scan
+ resp1 = client.post('/api/tasks', json={
+ 'type': 'scan',
+ 'path': str(test_dir),
+ 'recursive': True,
+ 'provider_id': 'local_fs',
+ 'root_id': '/'
+ })
+ assert resp1.status_code == 202
+
+ time.sleep(1) # Allow scan to process
+
+ # Get all scans
+ scans_resp = client.get('/api/scans')
+ assert scans_resp.status_code == 200
+ scans = scans_resp.get_json()
+ assert isinstance(scans, list)
+ assert len(scans) >= 1
+
+ # Verify scan appears in summary
+ found = any(s.get('path') == str(test_dir) for s in scans)
+ assert found, "Scan not found in unified history"
+
+
+def test_rclone_scan_with_options():
+ """Test that rclone-specific options are handled correctly."""
+ from scidk.app import create_app
+ app = create_app()
+ app.config['TESTING'] = True
+
+ with app.test_client() as client:
+ # Mock rclone scan with fast-list option
+ # Note: This will fail in test without actual rclone, but validates API contract
+ resp = client.post('/api/tasks', json={
+ 'type': 'scan',
+ 'path': 'dropbox:test',
+ 'recursive': True,
+ 'provider_id': 'rclone',
+ 'root_id': 'dropbox:',
+ 'fast_list': True
+ })
+
+ # Should accept the request format (will fail on execution without rclone)
+ assert resp.status_code in (202, 400, 500) # 202 if rclone available, error otherwise
+
+
+def test_snapshot_browser_after_scan(tmp_path: Path):
+ """Test viewing scan snapshot after completion."""
+ from scidk.app import create_app
+ app = create_app()
+ app.config['TESTING'] = True
+
+ test_dir = tmp_path / 'snapshot_test'
+ test_dir.mkdir()
+ (test_dir / 'data.csv').write_text('col1,col2\n1,2\n', encoding='utf-8')
+
+ with app.test_client() as client:
+ # Perform scan
+ scan_resp = client.post('/api/tasks', json={
+ 'type': 'scan',
+ 'path': str(test_dir),
+ 'recursive': True,
+ 'provider_id': 'local_fs',
+ 'root_id': '/'
+ })
+ assert scan_resp.status_code == 202
+ task_id = scan_resp.get_json()['task_id']
+
+ # Wait for completion
+ max_wait = 10
+ start_time = time.time()
+ scan_id = None
+
+ while time.time() - start_time < max_wait:
+ task_resp = client.get(f'/api/tasks/{task_id}')
+ task_data = task_resp.get_json()
+
+ if task_data['status'] == 'completed':
+ scan_id = task_data.get('scan_id')
+ break
+
+ time.sleep(0.5)
+
+ assert scan_id is not None, "Scan did not complete"
+
+ # Browse snapshot
+ snapshot_resp = client.get(f'/api/scans/{scan_id}/browse')
+ assert snapshot_resp.status_code == 200
+ snapshot_data = snapshot_resp.get_json()
+
+ assert 'entries' in snapshot_data
+ # Should find the data.csv file or parent folder
+ assert len(snapshot_data['entries']) >= 1
+
+
+def test_no_synchronous_scan_in_ui():
+ """Verify that synchronous /api/scan is NOT used by the Files page UI."""
+ from scidk.app import create_app
+ app = create_app()
+ app.config['TESTING'] = True
+
+ with app.test_client() as client:
+ resp = client.get('/datasets')
+ html = resp.data.decode('utf-8')
+
+ # Check that the JavaScript does NOT call /api/scan from provider panel
+ # (it should only use /api/tasks)
+ assert "'/api/scan'" not in html or html.count("'/api/scan'") <= 1
+ # Allow one mention in comments/strings, but not active code
+
+ # Verify /api/tasks is used instead
+ assert "'/api/tasks'" in html
+
+
+def test_current_location_display_updates():
+ """Test that the 'Current Location' panel updates when browsing."""
+ from scidk.app import create_app
+ app = create_app()
+ app.config['TESTING'] = True
+
+ with app.test_client() as client:
+ resp = client.get('/datasets')
+ html = resp.data.decode('utf-8')
+
+ # Check that current location display exists
+ assert 'prov-current-path' in html
+ assert 'Current Location:' in html
+
+ # Verify scan button is present and starts disabled
+ assert 'prov-scan-btn' in html
+ assert 'disabled' in html # Button should start disabled
+
+
+def test_scan_button_integration_with_background_form():
+ """Test that clicking scan button populates background scan form."""
+ from scidk.app import create_app
+ app = create_app()
+ app.config['TESTING'] = True
+
+ with app.test_client() as client:
+ resp = client.get('/datasets')
+ html = resp.data.decode('utf-8')
+
+ # Verify the scan button handler references background scan form elements
+ assert 'scan-path' in html # Background scan path input
+ assert 'scan-recursive' in html # Background scan recursive checkbox
+
+ # The JavaScript should populate these when scan button is clicked
+ # (Verified by manual testing and code inspection)
+
+
+def test_files_page_structure_consolidated():
+ """Verify that redundant sections have been removed/consolidated."""
+ from scidk.app import create_app
+ app = create_app()
+ app.config['TESTING'] = True
+
+ with app.test_client() as client:
+ resp = client.get('/datasets')
+ html = resp.data.decode('utf-8')
+ soup = BeautifulSoup(html, 'html.parser')
+
+ # Count h2 headings (main sections)
+ sections = soup.find_all('h2')
+ section_titles = [s.get_text() for s in sections]
+
+ # Should have core sections: Files, Snapshot browse, Scans Summary, Background Scan
+ assert 'Files' in section_titles
+ assert 'Snapshot (scanned) browse' in section_titles or 'Snapshot browse' in section_titles
+ assert 'Scans Summary' in section_titles
+ assert 'Start Background Scan' in section_titles
+
+ # Verify old sync scan form is gone
+ old_form = soup.find('form', id='prov-scan-form')
+ assert old_form is None, "Old synchronous scan form still present"
+
+
+def test_provider_selector_and_roots_load():
+ """Test that providers and roots load correctly."""
+ from scidk.app import create_app
+ app = create_app()
+ app.config['TESTING'] = True
+
+ with app.test_client() as client:
+ # Get providers
+ prov_resp = client.get('/api/providers')
+ assert prov_resp.status_code == 200
+ providers = prov_resp.get_json()
+ assert isinstance(providers, list)
+ assert len(providers) > 0
+
+ # Get roots for first provider
+ first_prov = providers[0]['id']
+ roots_resp = client.get(f'/api/provider_roots?provider_id={first_prov}')
+ assert roots_resp.status_code == 200
+ roots = roots_resp.get_json()
+ assert isinstance(roots, list)
diff --git a/tests/test_selective_scan_cache.py b/tests/test_selective_scan_cache.py
new file mode 100644
index 0000000..3f69836
--- /dev/null
+++ b/tests/test_selective_scan_cache.py
@@ -0,0 +1,220 @@
+"""
+Test selective scan cache integration using scan_items and directory_cache.
+"""
+import os
+import time
+from pathlib import Path
+import pytest
+
+
+def test_scan_populates_cache_tables(tmp_path: Path):
+ """Test that scans populate scan_items and directory_cache tables."""
+ from scidk.core import path_index_sqlite as pix
+ from scidk.core.migrations import migrate
+
+ # Setup test directory structure
+ (tmp_path / 'dir1').mkdir()
+ (tmp_path / 'dir1' / 'file1.txt').write_text('content1', encoding='utf-8')
+ (tmp_path / 'dir1' / 'file2.txt').write_text('content2', encoding='utf-8')
+ (tmp_path / 'dir2').mkdir()
+ (tmp_path / 'dir2' / 'file3.txt').write_text('content3', encoding='utf-8')
+
+ # Run a scan via the API
+ from scidk.app import create_app
+ app = create_app()
+ app.config['TESTING'] = True
+
+ with app.test_client() as client:
+ resp = client.post('/api/scan', json={
+ 'path': str(tmp_path),
+ 'recursive': True,
+ 'provider_id': 'local_fs'
+ })
+ assert resp.status_code == 200
+ data = resp.get_json()
+ assert data['status'] == 'ok'
+ scan_id = data['scan_id']
+
+ # Verify scan_items table is populated
+ conn = pix.connect()
+ migrate(conn)
+ cur = conn.cursor()
+ cur.execute("SELECT COUNT(*) FROM scan_items WHERE scan_id=?", (scan_id,))
+ count = cur.fetchone()[0]
+ assert count >= 5 # 2 dirs + 3 files
+
+ # Verify directory_cache table is populated
+ cur.execute("SELECT COUNT(*) FROM directory_cache WHERE scan_id=?", (scan_id,))
+ cache_count = cur.fetchone()[0]
+ assert cache_count >= 2 # At least tmp_path and dir1/dir2
+ conn.close()
+
+
+def test_rescan_uses_cache(tmp_path: Path):
+ """Test that rescanning unchanged directories uses cache and is faster."""
+ from scidk.app import create_app
+ app = create_app()
+ app.config['TESTING'] = True
+
+ # Create a larger directory structure
+ for i in range(5):
+ subdir = tmp_path / f'dir{i}'
+ subdir.mkdir()
+ for j in range(10):
+ (subdir / f'file{j}.txt').write_text(f'content{i}{j}', encoding='utf-8')
+
+ # First scan (cold)
+ with app.test_client() as client:
+ resp1 = client.post('/api/scan', json={
+ 'path': str(tmp_path),
+ 'recursive': True,
+ 'provider_id': 'local_fs'
+ })
+ assert resp1.status_code == 200
+ data1 = resp1.get_json()
+ duration1 = data1['duration_sec']
+ cache_stats1 = data1.get('cache_stats', {})
+ hits1 = cache_stats1.get('cache_hits', 0)
+
+ # First scan may or may not have cache hits depending on DB state
+ # (if test DB persists, there might be cached data from previous runs)
+
+ time.sleep(0.1) # Small delay to ensure different scan_id
+
+ # Second scan (warm - should use cache)
+ resp2 = client.post('/api/scan', json={
+ 'path': str(tmp_path),
+ 'recursive': True,
+ 'provider_id': 'local_fs'
+ })
+ assert resp2.status_code == 200
+ data2 = resp2.get_json()
+ duration2 = data2['duration_sec']
+ cache_stats2 = data2.get('cache_stats', {})
+ hits2 = cache_stats2.get('cache_hits', 0)
+
+ # Second scan should have at least as many cache hits as first
+ # (directory contents unchanged, so cache should be effective)
+ assert hits2 >= hits1
+
+
+def test_cache_detects_changes(tmp_path: Path):
+ """Test that cache correctly detects when directories change."""
+ from scidk.app import create_app
+ app = create_app()
+ app.config['TESTING'] = True
+
+ # Create initial structure
+ (tmp_path / 'dir1').mkdir()
+ (tmp_path / 'dir1' / 'file1.txt').write_text('content1', encoding='utf-8')
+
+ # First scan
+ with app.test_client() as client:
+ resp1 = client.post('/api/scan', json={
+ 'path': str(tmp_path),
+ 'recursive': True,
+ 'provider_id': 'local_fs'
+ })
+ assert resp1.status_code == 200
+
+ # Add a new file (changes directory)
+ (tmp_path / 'dir1' / 'file2.txt').write_text('content2', encoding='utf-8')
+ time.sleep(0.1)
+
+ # Second scan - cache should miss because directory changed
+ resp2 = client.post('/api/scan', json={
+ 'path': str(tmp_path),
+ 'recursive': True,
+ 'provider_id': 'local_fs'
+ })
+ assert resp2.status_code == 200
+ data2 = resp2.get_json()
+
+ # Should detect the new file
+ assert data2['scanned'] >= 2
+
+
+def test_cache_can_be_disabled(tmp_path: Path):
+ """Test that cache can be disabled via environment variable."""
+ from scidk.app import create_app
+
+ # Create test structure
+ (tmp_path / 'file.txt').write_text('content', encoding='utf-8')
+
+ # Disable cache
+ original = os.environ.get('SCIDK_CACHE_SCAN')
+ try:
+ os.environ['SCIDK_CACHE_SCAN'] = '0'
+ app = create_app()
+ app.config['TESTING'] = True
+
+ with app.test_client() as client:
+ # First scan
+ resp1 = client.post('/api/scan', json={
+ 'path': str(tmp_path),
+ 'recursive': True,
+ 'provider_id': 'local_fs'
+ })
+ assert resp1.status_code == 200
+ data1 = resp1.get_json()
+ cache_stats1 = data1.get('cache_stats', {})
+
+ # Cache should be disabled
+ assert cache_stats1.get('enabled') is False
+
+ time.sleep(0.1)
+
+ # Second scan
+ resp2 = client.post('/api/scan', json={
+ 'path': str(tmp_path),
+ 'recursive': True,
+ 'provider_id': 'local_fs'
+ })
+ assert resp2.status_code == 200
+ data2 = resp2.get_json()
+ cache_stats2 = data2.get('cache_stats', {})
+
+ # Cache should still be disabled
+ assert cache_stats2.get('enabled') is False
+ finally:
+ if original is not None:
+ os.environ['SCIDK_CACHE_SCAN'] = original
+ elif 'SCIDK_CACHE_SCAN' in os.environ:
+ del os.environ['SCIDK_CACHE_SCAN']
+
+
+def test_cache_helpers():
+ """Test cache helper functions in path_index_sqlite."""
+ import hashlib
+ import time
+ from scidk.core import path_index_sqlite as pix
+ from scidk.core.migrations import migrate
+
+ # Use unique scan_id to avoid conflicts
+ scan_id = f"test_scan_{hashlib.sha1(str(time.time()).encode()).hexdigest()[:12]}"
+
+ # Test record_scan_items
+ rows = [
+ ('/tmp/file1.txt', 'file', 100, 1234567890.0, '.txt', 'text/plain', None, 'hash1', None),
+ ('/tmp/file2.txt', 'file', 200, 1234567891.0, '.txt', 'text/plain', None, 'hash2', None),
+ ]
+ inserted = pix.record_scan_items(scan_id, rows)
+ assert inserted == 2
+
+ # Test cache_directory_listing
+ pix.cache_directory_listing(scan_id, '/tmp', ['file1.txt', 'file2.txt'])
+
+ # Test get_cached_directory
+ cached = pix.get_cached_directory(scan_id, '/tmp')
+ assert cached == ['file1.txt', 'file2.txt']
+
+ # Test get_previous_scan_for_path
+ prev = pix.get_previous_scan_for_path('/tmp/file1.txt')
+ assert prev == scan_id
+
+ # Test get_scan_item
+ item = pix.get_scan_item(scan_id, '/tmp/file1.txt')
+ assert item is not None
+ assert item['path'] == '/tmp/file1.txt'
+ assert item['type'] == 'file'
+ assert item['size'] == 100