/ justapi / crates / justapi-py / test_hot_reload.py
swadhinbiswas test(py): add comprehensive test suite for Python bindings · 7/8/2026
f693697
84 lines · 2.3 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
import pytest
import os
import time
import json
import threading
import requests
from justapi import JustAPIApp
import asyncio

def test_hot_reload():
    config_path = "test_gateway.json"
    
    initial_config = {
        "routes": {
            "/api/proxy": {
                "GET": {
                    "upstream": "http://v1.upstream.com"
                }
            }
        }
    }
    with open(config_path, "w") as f:
        json.dump(initial_config, f)
        
    app = JustAPIApp()
    app.enable_gateway(config_path)

    @app.get("/api/local")
    def local_route():
        return {"msg": "local"}

    def run_server():
        app.run("127.0.0.1:9199")

    t = threading.Thread(target=run_server, daemon=True)
    t.start()
    time.sleep(1) # wait for server to start

    # 1. Test Local
    resp = requests.get("http://127.0.0.1:9199/api/local")
    assert resp.status_code == 200
    assert resp.json() == {"msg": "local"}
    
    # 2. Test initial Gateway Proxy
    resp = requests.get("http://127.0.0.1:9199/api/proxy")
    assert resp.status_code == 200
    assert resp.headers["x-justapi-gateway"] == "hot-reloaded"
    assert resp.content == b"Proxied to: http://v1.upstream.com"
    
    # 3. Modify Gateway config (HOT RELOAD)
    updated_config = {
        "routes": {
            "/api/proxy": {
                "GET": {
                    "upstream": "http://v2.upstream.com" # Updated!
                }
            },
            "/api/new-proxy": {
                "GET": {
                    "upstream": "http://v3.new.upstream.com"
                }
            }
        }
    }
    with open(config_path, "w") as f:
        json.dump(updated_config, f)
        
    time.sleep(1) # wait for file watcher to pick up and debounce
    
    # 4. Test updated Gateway Proxy
    resp = requests.get("http://127.0.0.1:9199/api/proxy")
    assert resp.status_code == 200
    assert resp.headers["x-justapi-gateway"] == "hot-reloaded"
    assert resp.content == b"Proxied to: http://v2.upstream.com"
    
    resp = requests.get("http://127.0.0.1:9199/api/new-proxy")
    assert resp.status_code == 200
    assert resp.content == b"Proxied to: http://v3.new.upstream.com"
    
    print("Test Passed: Hot Reloading Works Successfully!")
    
if __name__ == "__main__":
    test_hot_reload()