pye

Author: manatlan

examples/pye/server.py

@@ -25,6 +25,7 @@
 logger = logging.getLogger("pye")
 
 APP_TTL = 10 * 60  # 10 minutes
+AUTO_INDEX = False  # Set to False to disable directory index listing
 
 
 def get_app_mtime(p: Path) -> float:
@@ -242,6 +243,27 @@ def _unload_expired_apps(self):
                 self.router_logger.info(f"TTL expired for htag app '{name}', unloading...")
                 info.app = None
 
+    async def _serve_htag_app(self, name: str, info: AppInfo, scope: Scope, receive: Receive, send: Send, new_path: str) -> bool:
+        """Loads and serves an htag application."""
+        info.last_accessed = time.time()
+        mtime = get_app_mtime(info.file)
+        
+        if info.app is None or (scope["type"] == "http" and mtime > info.mtime):
+            action = "Reloading" if info.app else "Loading"
+            self.router_logger.info(f"{action} htag app: {name}")
+            app_class = load_htag_app(info.mod_path)
+            if app_class:
+                info.app, info.mtime = WebApp(app_class).app, mtime
+            else:
+                return False
+
+        if info.app:
+            scope["path"] = new_path
+            scope["root_path"] = scope.get("root_path", "") + "/" + name
+            await info.app(scope, receive, send)
+            return True
+        return False
+
     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
         if scope["type"] not in ["http", "websocket"]: return
         if scope["type"] == "http": self._unload_expired_apps()
@@ -255,20 +277,8 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
             for name in sorted(self.apps.keys(), key=len, reverse=True):
                 if rel_path == name or rel_path.startswith(name + "/"):
                     info = self.apps[name]
-                    info.last_accessed = time.time()
-                    mtime = get_app_mtime(info.file)
-                    
-                    if info.app is None or (scope["type"] == "http" and mtime > info.mtime):
-                        action = "Reloading" if info.app else "Loading"
-                        self.router_logger.info(f"{action} htag app: {name}")
-                        app_class = load_htag_app(info.mod_path)
-                        if app_class:
-                            info.app, info.mtime = WebApp(app_class).app, mtime
-
-                    if info.app:
-                        scope["path"] = path[len("/" + name):] or "/"
-                        scope["root_path"] = scope.get("root_path", "") + "/" + name
-                        await info.app(scope, receive, send)
+                    new_path = path[len("/" + name):] or "/"
+                    if await self._serve_htag_app(name, info, scope, receive, send, new_path):
                         return
 
         # 2. Directory or Index
@@ -284,8 +294,24 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
                 if rel_path and not path.endswith("/"):
                     await HTMLResponse("", status_code=301, headers={"Location": path + "/"})(scope, receive, send)
                 else:
-                    self.router_logger.info(f"Serving directory index: {path}")
-                    await HTMLResponse(IndexGenerator.generate(self.apps_dir, self.apps, rel_path))(scope, receive, send)
+                    if AUTO_INDEX:
+                        self.router_logger.info(f"Serving directory index: {path}")
+                        await HTMLResponse(IndexGenerator.generate(self.apps_dir, self.apps, rel_path))(scope, receive, send)
+                    else:
+                        index_name = f"{rel_path}/index" if rel_path else "index"
+                        if index_name in self.apps:
+                            if await self._serve_htag_app(index_name, self.apps[index_name], scope, receive, send, "/"):
+                                return
+
+                        index_file = full_path / "index.py"
+                        if index_file.is_file():
+                            self.router_logger.info(f"Running default index script: {index_file}")
+                            response = await ScriptRunner.run(index_file, scope)
+                            await response(scope, receive, send)
+                            return
+
+                        self.router_logger.warning(f"No entrypoint found and AUTO_INDEX is False: {path}")
+                        await HTMLResponse("No default entrypoint", status_code=400)(scope, receive, send)
             return
 
         # 3. Static or Script
@@ -317,5 +343,6 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
 
 if __name__ == "__main__":
     import uvicorn
+    AUTO_INDEX = True  # Enable indexing for demo
     uvicorn.run(app, host="0.0.0.0", port=8000)
 

examples/pye/test_server.py

@@ -66,15 +66,21 @@ def client(temp_apps_dir):
     return TestClient(app)
 
 def test_root_index(client):
-    response = client.get("/")
-    assert response.status_code == 200
-    assert "text/html" in response.headers["content-type"]
-    assert "Root Index" in response.text
-    # Check that our test files are listed
-    assert "test.txt" in response.text
-    assert "myscript.py" in response.text
-    assert "testapp" in response.text
-    assert "myfolder/" in response.text
+    import server
+    original_val = server.AUTO_INDEX
+    try:
+        server.AUTO_INDEX = True
+        response = client.get("/")
+        assert response.status_code == 200
+        assert "text/html" in response.headers["content-type"]
+        assert "Root Index" in response.text
+        # Check that our test files are listed
+        assert "test.txt" in response.text
+        assert "myscript.py" in response.text
+        assert "testapp" in response.text
+        assert "myfolder/" in response.text
+    finally:
+        server.AUTO_INDEX = original_val
 
 def test_static_file(client, temp_apps_dir):
     response = client.get("/test.txt")
@@ -197,18 +203,130 @@ def test_script_timeout(client, temp_apps_dir):
     assert "done" in response.text
 
 def test_navigation_depth(client, temp_apps_dir):
-    # Test deeper nesting and "Parent Directory" links
-    sub = temp_apps_dir / "folder" / "sub"
-    sub.mkdir(parents=True)
-    (sub / "file.txt").write_text("deep")
+    import server
+    original_val = server.AUTO_INDEX
+    try:
+        server.AUTO_INDEX = True
+        # Test deeper nesting and "Parent Directory" links
+        sub = temp_apps_dir / "folder" / "sub"
+        sub.mkdir(parents=True)
+        (sub / "file.txt").write_text("deep")
+        
+        # Check /folder/ index
+        resp1 = client.get("/folder/")
+        assert 'href="/folder/sub/"' in resp1.text
+        assert 'href="/"' in resp1.text # Parent from /folder/ is /
+        
+        # Check /folder/sub/ index
+        resp2 = client.get("/folder/sub/")
+        assert 'href="/folder/sub/file.txt"' in resp2.text
+        assert 'href="/folder/"' in resp2.text # Parent from /folder/sub/ is /folder/
+    finally:
+        server.AUTO_INDEX = original_val
+
+def test_auto_index_disabled(temp_apps_dir):
+    import server
+    # Backup original
+    original_val = server.AUTO_INDEX
+    try:
+        server.AUTO_INDEX = False
+        router = server.DynamicHtagApps(temp_apps_dir)
+        from starlette.testclient import TestClient
+        client = TestClient(router)
+        
+        response = client.get("/")
+        assert response.status_code == 400
+        assert response.text == "No default entrypoint"
+    finally:
+        server.AUTO_INDEX = original_val
+
+def test_default_index_app(client, temp_apps_dir):
+    # Create an index htag app in a folder
+    idx_dir = temp_apps_dir / "hasindex"
+    idx_dir.mkdir()
+    app_content = """from htag import Tag
+class App(Tag.App):
+    def init(self):
+        self += Tag.h1("I am the index app")
+"""
+    (idx_dir / "index.py").write_text(app_content)
 
-    # Check /folder/ index
-    resp1 = client.get("/folder/")
-    assert 'href="/folder/sub/"' in resp1.text
-    assert 'href="/"' in resp1.text # Parent from /folder/ is /
+    import server
+    original_val = server.AUTO_INDEX
+    try:
+        server.AUTO_INDEX = False # Must be False to trigger index check
+        router = server.DynamicHtagApps(temp_apps_dir)
+        from starlette.testclient import TestClient
+        client = TestClient(router)
+
+        response = client.get("/hasindex/")
+        assert response.status_code == 200
+        assert "I am the index app" in response.text
+    finally:
+        server.AUTO_INDEX = original_val
+
+def test_default_index_script(client, temp_apps_dir):
+    # Create an index script in a folder
+    idx_dir = temp_apps_dir / "hasidxscript"
+    idx_dir.mkdir()
+    (idx_dir / "index.py").write_text("print('I am the index script')")
 
-    # Check /folder/sub/ index
-    resp2 = client.get("/folder/sub/")
-    assert 'href="/folder/sub/file.txt"' in resp2.text
-    assert 'href="/folder/"' in resp2.text # Parent from /folder/sub/ is /folder/
+    import server
+    original_val = server.AUTO_INDEX
+    try:
+        server.AUTO_INDEX = False # Must be False to trigger index check
+        router = server.DynamicHtagApps(temp_apps_dir)
+        from starlette.testclient import TestClient
+        client = TestClient(router)
+
+        response = client.get("/hasidxscript/")
+        assert response.status_code == 200
+        assert "I am the index script" in response.text
+    finally:
+        server.AUTO_INDEX = original_val
+
+def test_auto_index_enabled_ignores_index_file(client, temp_apps_dir):
+    # If AUTO_INDEX is True, it should list the directory even if index.py exists
+    idx_dir = temp_apps_dir / "index_enabled"
+    idx_dir.mkdir()
+    (idx_dir / "index.py").write_text("print('I should not run')")
+    (idx_dir / "other.txt").write_text("other")
+
+    import server
+    original_val = server.AUTO_INDEX
+    try:
+        server.AUTO_INDEX = True 
+        router = server.DynamicHtagApps(temp_apps_dir)
+        from starlette.testclient import TestClient
+        client = TestClient(router)
+
+        response = client.get("/index_enabled/")
+        assert response.status_code == 200
+        # Should be a listing
+        assert "Index of index_enabled/" in response.text
+        assert "index.py" in response.text
+        assert "other.txt" in response.text
+        assert "I should not run" not in response.text
+    finally:
+        server.AUTO_INDEX = original_val
+
+def test_auto_index_disabled_in_subfolder(client, temp_apps_dir):
+    # Verify 400 even in nested folders if AUTO_INDEX=False and no index
+    sub = temp_apps_dir / "mysub" / "deep"
+    sub.mkdir(parents=True)
+    (sub / "hidden.txt").write_text("secret")
+
+    import server
+    original_val = server.AUTO_INDEX
+    try:
+        server.AUTO_INDEX = False
+        router = server.DynamicHtagApps(temp_apps_dir)
+        from starlette.testclient import TestClient
+        client = TestClient(router)
+        
+        response = client.get("/mysub/deep/")
+        assert response.status_code == 400
+        assert response.text == "No default entrypoint"
+    finally:
+        server.AUTO_INDEX = original_val