scratch/test_endpoints.py

import urllib.request
import urllib.parse
import json
import numpy as np

def test_get_benchmarks():
    print("Testing GET /api/benchmarks ...")
    try:
        response = urllib.request.urlopen("http://localhost:7860/api/benchmarks")
        data = json.loads(response.read().decode())
        print("  SUCCESS!")
        print("  Benchmarks returned:", len(data), "items")
        for item in data[:1]:
            print("  Example benchmark model name:", item["model"])
    except Exception as e:
        print("  FAILED:", e)

def test_search_text():
    print("Testing POST /api/search-text ...")
    try:
        url = "http://localhost:7860/api/search-text"
        data = urllib.parse.urlencode({
            "text_query": "forest",
            "k": 3,
            "level": "level4",
            "query_modality": "optical",
            "lat": 20.5937,
            "lon": 78.9629,
            "radius_km": 50.0
        }).encode()
        
        req = urllib.request.Request(url, data=data, method="POST")
        response = urllib.request.urlopen(req)
        res_data = json.loads(response.read().decode())
        print("  SUCCESS!")
        print("  Query time (ms):", res_data["query_time_ms"])
        print("  Results count:", len(res_data["results"]))
        for i, item in enumerate(res_data["results"]):
            print(f"  Result #{i+1}: class={item['class']}, modality={item['modality']}, score={item['score']:.4f}")
            print(f"    lat={item['lat']:.4f}, lon={item['lon']:.4f}, distance={item['distance_km']:.2f} km")
            print(f"    h3_cell={item['h3_cell']}, h3_boundary={len(item['h3_boundary'])} points")
    except Exception as e:
        print("  FAILED:", e)

def test_render_bands():
    print("Testing GET /api/render-bands ...")
    try:
        # Get a sample path from metadata
        res = urllib.request.urlopen("http://localhost:7860/api/search-text", data=urllib.parse.urlencode({
            "text_query": "river", "k": 1, "level": "level1", "query_modality": "multispectral"
        }).encode())
        res_data = json.loads(res.read().decode())
        if not res_data["results"]:
            print("  SKIP (no multispectral file to render)")
            return
            
        sample_path = res_data["results"][0]["original_path"]
        print("  Selected sample path:", sample_path)
        
        # Request FCC render
        url = f"http://localhost:7860/api/render-bands?path={urllib.parse.quote(sample_path)}&bands=FCC"
        response = urllib.request.urlopen(url)
        img_bytes = response.read()
        print("  SUCCESS!")
        print("  Rendered FCC bytes size:", len(img_bytes), "bytes")
        
        # Request RGB render
        url = f"http://localhost:7860/api/render-bands?path={urllib.parse.quote(sample_path)}&bands=RGB"
        response = urllib.request.urlopen(url)
        img_bytes = response.read()
        print("  SUCCESS!")
        print("  Rendered RGB bytes size:", len(img_bytes), "bytes")
    except Exception as e:
        print("  FAILED:", e)

def test_spectral_signature():
    print("Testing GET /api/spectral-signature ...")
    try:
        res = urllib.request.urlopen("http://localhost:7860/api/search-text", data=urllib.parse.urlencode({
            "text_query": "river", "k": 1, "level": "level1", "query_modality": "multispectral"
        }).encode())
        res_data = json.loads(res.read().decode())
        if not res_data["results"]:
            print("  SKIP (no multispectral file to plot)")
            return
            
        sample_path = res_data["results"][0]["original_path"]
        url = f"http://localhost:7860/api/spectral-signature?path={urllib.parse.quote(sample_path)}"
        response = urllib.request.urlopen(url)
        res_data = json.loads(response.read().decode())
        print("  SUCCESS!")
        print("  Bands count:", len(res_data["reflectance"]))
        print("  Bands list values:", [round(v, 3) for v in res_data["reflectance"]])
    except Exception as e:
        print("  FAILED:", e)

def test_get_root():
    print("Testing GET / (Custom UI Root)...")
    try:
        response = urllib.request.urlopen("http://localhost:7860/")
        content = response.read().decode()
        if "<title>SatFetch" in content:
            print("  SUCCESS!")
            print("  Custom UI served at root index page correctly!")
        else:
            print("  FAILED: Title tag not found in root response content.")
    except Exception as e:
        print("  FAILED:", e)

def test_search_file_upload():
    print("Testing POST /api/search (File Upload with non-standard dimensions)...")
    try:
        import tifffile
        import tempfile
        import os
        
        dummy_data = np.random.randint(0, 255, (228, 238, 3), dtype=np.uint8)
        with tempfile.NamedTemporaryFile(suffix=".tif", delete=False) as tmp:
            tifffile.imwrite(tmp.name, dummy_data)
            tmp_path = tmp.name
            
        boundary = "----WebKitFormBoundary7MA4YWxkTrZu0gW"
        parts = []
        
        parts.append(f"--{boundary}")
        parts.append('Content-Disposition: form-data; name="file"; filename="upload_test.tif"')
        parts.append('Content-Type: image/tiff')
        parts.append('')
        with open(tmp_path, "rb") as f:
            parts.append(f.read())
            
        parts.append(f"--{boundary}")
        parts.append('Content-Disposition: form-data; name="query_modality"')
        parts.append('')
        parts.append('optical')
        
        parts.append(f"--{boundary}")
        parts.append('Content-Disposition: form-data; name="k"')
        parts.append('')
        parts.append('3')
        
        parts.append(f"--{boundary}")
        parts.append('Content-Disposition: form-data; name="level"')
        parts.append('')
        parts.append('level2')
        
        parts.append(f"--{boundary}--")
        parts.append('')
        
        body = b""
        for p in parts:
            if isinstance(p, str):
                body += p.encode("utf-8") + b"\r\n"
            else:
                body += p + b"\r\n"
                
        req = urllib.request.Request(
            "http://localhost:7860/api/search",
            data=body,
            headers={
                "Content-Type": f"multipart/form-data; boundary={boundary}",
                "Content-Length": str(len(body))
            }
        )
        
        response = urllib.request.urlopen(req)
        res_data = json.loads(response.read().decode())
        print("  SUCCESS!")
        print("  Results count:", len(res_data["results"]))
        print("  Query time (ms):", res_data["query_time_ms"])
        
        os.remove(tmp_path)
    except Exception as e:
        print("  FAILED:", e)

if __name__ == "__main__":
    print("Executing programatic endpoint tests on active server...")
    test_get_root()
    print("-" * 50)
    test_get_benchmarks()
    print("-" * 50)
    test_search_text()
    print("-" * 50)
    test_render_bands()
    print("-" * 50)
    test_spectral_signature()
    print("-" * 50)
    test_search_file_upload()