diff --git a/.vscode/launch.json b/.vscode/launch.json index 55ba1bd..951e27e 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -1,14 +1,14 @@ { "version": "0.2.0", "configurations": [ - { "name": "Python Debugger: Current File with Arguments", "type": "debugpy", "request": "launch", "program": "${file}", "console": "integratedTerminal", - "args": "${command:pickArgs}" - } + "args": "--port=53535 --root-path=/proxy/53535 --navigate *.zip" + }, + ] } \ No newline at end of file diff --git a/main.py b/main.py index 4d24ac2..a9f8275 100644 --- a/main.py +++ b/main.py @@ -23,6 +23,7 @@ file_mapping: dict[str, str] = {} indexers = [] sorted_hashes: list[str] = [] navigate_enabled = False +root_path: str = "" class FileIndexer: @@ -166,8 +167,13 @@ INDEXER_MAP = {".zip": ZipFileIndexer} def initialize_server(args: argparse.Namespace): """Initialize the server with directory or glob indexing""" - global file_mapping, indexers, sorted_hashes, navigate_enabled + global file_mapping, indexers, sorted_hashes, navigate_enabled, root_path navigate_enabled = args.navigate + root_path = getattr(args, "root_path", None) or "" + if root_path: + # Ensure root_path starts with / but not ends with / + root_path = "/" + root_path.strip("/") + app.root_path = root_path src_path = Path(args.source) @@ -259,7 +265,7 @@ def _build_url( file_hash: str, order: str | None = None, delay: int | None = None ) -> str: """Build a URL with optional order/delay query parameters.""" - base = f"/{file_hash}" + base = f"{root_path}/{file_hash}" if order is not None and delay is not None: return f"{base}?order={order}&delay={delay}" return base @@ -282,7 +288,7 @@ async def root( ): _raise_unauthorized() - return RedirectResponse(url=_build_url(random_hash, order, delay)) + return RedirectResponse(url=_build_url(random_hash, order, delay), status_code=307) def _get_navigation_data(file_hash: str, order: str | None = None): @@ -355,8 +361,8 @@ def _render_page( prev_path, order=current_order, delay=current_delay ) else: - next_url = f"/navigate/{next_path}" - prev_url = f"/navigate/{prev_path}" + next_url = f"{root_path}/navigate/{next_path}" + prev_url = f"{root_path}/navigate/{prev_path}" else: if current_order is not None: next_url = _build_url( @@ -370,8 +376,12 @@ def _render_page( delay=current_delay, ) else: - next_url = "/{next_hash}".format(next_hash=navigation_data["next_hash"]) - prev_url = "/{prev_hash}".format(prev_hash=navigation_data["prev_hash"]) + next_url = "{root_path}/{next_hash}".format( + root_path=root_path, next_hash=navigation_data["next_hash"] + ) + prev_url = "{root_path}/{prev_hash}".format( + root_path=root_path, prev_hash=navigation_data["prev_hash"] + ) # Build the toggle URL (alternate view mode) file_path = navigation_data.get("filename", "") @@ -404,7 +414,9 @@ def _render_page( sidebar_class = "" content = template.substitute( - img_url="/api/{file_hash}/data".format(file_hash=navigation_data["file_hash"]), + img_url="{root_path}/api/{file_hash}/data".format( + root_path=root_path, file_hash=navigation_data["file_hash"] + ), image_click_url=image_click_url or _get_random_hash(), next_url=next_url, prev_url=prev_url, @@ -465,7 +477,9 @@ async def hash_page( image_click_url = _build_url(file_hash) # Create pause button to stop auto-refresh - pause_button = f'' + pause_button = ( + f'' + ) return _render_page( navigation_data, @@ -478,7 +492,7 @@ async def hash_page( else: # Browse mode play_button = ( - f'' ) return _render_page( @@ -606,7 +620,7 @@ def _build_navigate_url( path: str, order: str | None = None, delay: int | None = None ) -> str: """Build a /navigate URL with optional order/delay query parameters.""" - base = f"/navigate/{path}" + base = f"{root_path}/navigate/{path}" if order is not None and delay is not None: return f"{base}?order={order}&delay={delay}" return base @@ -704,24 +718,15 @@ def _render_folder_index_html( all_paths = _collect_zip_paths() lines: list[str] = [] - # Build root-level folders and files - root_folders: set[str] = set() - root_files: set[str] = set() - for p in all_paths: - if "/" in p: - root_folders.add(p.split("/", 1)[0]) - else: - root_files.add(p) - # Breadcrumb lines.append("") def _render_folder(folder_prefix: str, depth: int = 0) -> list[str]: @@ -744,7 +749,7 @@ def _render_folder_index_html( folder_path = f"{prefix}{folder}/" if prefix else f"{folder}/" result.append( f'
' - f'{indent}📁 {folder}' + f'{indent}📁 {folder}' f"
" ) result.extend(_render_folder(folder_path, depth + 1)) @@ -758,7 +763,7 @@ def _render_folder_index_html( href_params = f"?order={current_order}&delay={current_delay}" result.append( f'
' - f'{indent}📄 {file}' + f'{indent}📄 {file}' f"
" ) @@ -884,9 +889,7 @@ async def navigate_page( image_click_url = _build_navigate_url(path) # Create pause button to stop auto-refresh - pause_button = ( - f'' - ) + pause_button = f'' return _render_page( navigation_data, @@ -900,12 +903,12 @@ async def navigate_page( else: # Browse mode play_button = ( - f'' ) random_path = _get_random_navigate_path() image_click_url = ( - f"/navigate/{random_path}" if random_path else _get_random_hash() + f"{root_path}/navigate/{random_path}" if random_path else _get_random_hash() ) return _render_page( navigation_data, @@ -934,6 +937,12 @@ if __name__ == "__main__": action="store_true", help="Enable path-based navigation (/navigate/{path})", ) + parser.add_argument( + "--root-path", + type=str, + default=None, + help="URL path prefix (e.g., /images for serving at example.com/images)", + ) args = parser.parse_args() initialize_server(args) diff --git a/tests/conftest.py b/tests/conftest.py index 1335adb..cb7a6c1 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -18,6 +18,8 @@ def _reset_state() -> None: main.indexers.clear() main.sorted_hashes.clear() main.navigate_enabled = False + main.root_path = "" + main.app.root_path = "" @pytest.fixture(autouse=True) @@ -86,6 +88,7 @@ def args_directory(sample_files: dict[str, Path], tmp_path: Path) -> argparse.Na port=0, salt="test-salt", navigate=False, + root_path=None, ) @@ -98,6 +101,7 @@ def args_zip(sample_zip: dict[str, Path], tmp_path: Path) -> argparse.Namespace: port=0, salt="test-salt", navigate=False, + root_path=None, ) @@ -112,6 +116,7 @@ def args_zip_navigate( port=0, salt="test-salt", navigate=True, + root_path=None, ) @@ -190,6 +195,7 @@ def args_encrypted_zip( port=0, salt="test-salt", navigate=False, + root_path=None, ) diff --git a/tests/test_root_path.py b/tests/test_root_path.py new file mode 100644 index 0000000..78e1ad3 --- /dev/null +++ b/tests/test_root_path.py @@ -0,0 +1,181 @@ +"""Tests for --root-path URL prefix support.""" + +import argparse +from collections.abc import Generator +from pathlib import Path + +import pytest +from httpx import ASGITransport, AsyncClient + +import main + + +@pytest.fixture +def args_directory_with_root( + sample_files: dict[str, Path], tmp_path: Path +) -> argparse.Namespace: + """Argparse namespace with root_path set.""" + return argparse.Namespace( + source=str(tmp_path), + host="127.0.0.1", + port=0, + salt="test-salt", + navigate=False, + root_path="/images", + ) + + +@pytest.fixture +def initialized_dir_with_root(args_directory_with_root: argparse.Namespace) -> None: + """Initialize the server with sample directory files and root_path.""" + main.initialize_server(args_directory_with_root) + + +@pytest.fixture +async def client_dir_with_root( + initialized_dir_with_root: None, +) -> Generator[AsyncClient]: + """Async HTTP client against the app initialized with root_path.""" + transport = ASGITransport(app=main.app) + async with AsyncClient(transport=transport, base_url="http://test") as ac: + yield ac + + +class TestRootPathGlobal: + """Tests that root_path is set correctly on initialization.""" + + def test_root_path_set_from_args(self, args_directory_with_root): + """root_path global is set from args.""" + main.initialize_server(args_directory_with_root) + assert main.root_path == "/images" + + def test_root_path_normalized(self, sample_files, tmp_path): + """root_path is normalized (leading/trailing slashes handled).""" + args = argparse.Namespace( + source=str(tmp_path), + host="127.0.0.1", + port=0, + salt="test-salt", + navigate=False, + root_path="images/", + ) + main.initialize_server(args) + assert main.root_path == "/images" + + def test_root_path_empty_when_not_provided(self, args_directory): + """root_path is empty string when not provided.""" + main.initialize_server(args_directory) + assert main.root_path == "" + + def test_app_root_path_set(self, args_directory_with_root): + """FastAPI app.root_path is set for proper URL generation.""" + main.initialize_server(args_directory_with_root) + assert main.app.root_path == "/images" + + +class TestRootPathURLBuilders: + """Tests that URL builder functions include root_path.""" + + def test_build_url_includes_root_path(self, initialized_dir_with_root): + """_build_url includes root_path prefix.""" + url = main._build_url("abc123") + assert url == "/images/abc123" + + def test_build_url_with_params_includes_root_path(self, initialized_dir_with_root): + """_build_url with order/delay includes root_path prefix.""" + url = main._build_url("abc123", order="next", delay=5) + assert url == "/images/abc123?order=next&delay=5" + + def test_build_navigate_url_includes_root_path(self, initialized_dir_with_root): + """_build_navigate_url includes root_path prefix.""" + url = main._build_navigate_url("folder/file.txt") + assert url == "/images/navigate/folder/file.txt" + + def test_build_navigate_url_with_params_includes_root_path( + self, initialized_dir_with_root + ): + """_build_navigate_url with order/delay includes root_path prefix.""" + url = main._build_navigate_url("folder/file.txt", order="random", delay=3) + assert url == "/images/navigate/folder/file.txt?order=random&delay=3" + + def test_build_url_no_root_when_empty(self, initialized_dir): + """_build_url has no prefix when root_path is empty.""" + url = main._build_url("abc123") + assert url == "/abc123" + + +class TestRootPathEndpoints: + """Tests that endpoint responses include root_path in URLs.""" + + async def test_root_redirect_includes_root_path( + self, client_dir_with_root: AsyncClient + ): + """Root redirect location includes root_path.""" + response = await client_dir_with_root.get("/", follow_redirects=False) + assert response.status_code == 307 + location = response.headers["location"] + assert location.startswith("/images/") + + async def test_root_redirect_with_params_includes_root_path( + self, client_dir_with_root: AsyncClient + ): + """Root redirect with order/delay includes root_path.""" + response = await client_dir_with_root.get( + "/", params={"order": "next", "delay": 5}, follow_redirects=False + ) + assert response.status_code == 307 + location = response.headers["location"] + assert location.startswith("/images/") + assert "order=next" in location + assert "delay=5" in location + + async def test_hash_page_image_url_includes_root_path( + self, client_dir_with_root: AsyncClient + ): + """Hash page img_url includes root_path.""" + file_hash = list(main.file_mapping.keys())[0] + response = await client_dir_with_root.get(f"/{file_hash}") + assert response.status_code == 200 + assert "/images/api/" in response.text + + async def test_hash_page_nav_urls_include_root_path( + self, client_dir_with_root: AsyncClient + ): + """Hash page prev/next URLs include root_path.""" + file_hash = list(main.file_mapping.keys())[0] + response = await client_dir_with_root.get(f"/{file_hash}") + assert response.status_code == 200 + # Check that chevron links use root_path + import re + + # Find href values on chevron elements + chevrons = re.findall( + r'class="chevron (left|right)"[^>]*href="([^"]*)"', response.text + ) + for direction, href in chevrons: + assert href.startswith( + "/images/" + ), f"Chevron {direction} href {href} missing root_path" + + async def test_hash_page_play_button_includes_root_path( + self, client_dir_with_root: AsyncClient + ): + """Hash page play button URL includes root_path.""" + file_hash = list(main.file_mapping.keys())[0] + response = await client_dir_with_root.get(f"/{file_hash}") + assert response.status_code == 200 + # Play button should have root_path in its href + assert "/images/" in response.text + + async def test_hash_page_refresh_includes_root_path( + self, client_dir_with_root: AsyncClient + ): + """Hash page in refresh mode includes root_path in meta refresh.""" + file_hash = list(main.file_mapping.keys())[0] + response = await client_dir_with_root.get( + f"/{file_hash}", params={"order": "next", "delay": 5} + ) + assert response.status_code == 200 + assert 'http-equiv="refresh"' in response.text + # The refresh URL should include root_path + assert "/images/" in response.text