Files
Pier_v1/RAG/design-model-box-sample_2/deployment/test_subpath_gateway.py
T
ahnyongjooandClaude Sonnet 5.5 964a841de1 chore(rebar): Pier_T_rebar 작업 폴더 초기 구성 및 외부 의존 파일 통합
- Code\rebar_pipeline, review_pipeline, pier-program을 정션에서 실제 폴더로 전환
- Code\rhino_mcp 및 CLAUDE_rhino배경.md 추가, .mcp.json은 PYTHONPATH로 새 위치 사용
- pier-program 내부 .git 제거(이력은 backup\pier-program_20261028.bundle에 보관)
- worktree 가이드 반영(CLAUDE.local.md, HANDOFF.md)
- D:\Rhino-MCP 경로 참조를 새 위치로 치환

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-28 11:20:18 +09:00

67 lines
3.0 KiB
Python

import http.client
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import threading
import unittest
import subpath_gateway as gateway
class Backend(BaseHTTPRequestHandler):
def do_GET(self):
if self.path=='/api/status':
data=b'{"ok":true,"asset":"/api/uploads/file","unchanged":"/tmp/model.json"}'
mime='application/json'
elif self.path=='/src/app.js':
data=b"fetch('/api/status'); frame.src=`/dxf_rasterizer.html?file=x`;"
mime='application/javascript'
elif self.path=='/api/uploads/file':data=b'\x00\xff\x01';mime='application/octet-stream'
else:data=b'<link href="styles.css">';mime='text/html'
self.send_response(200);self.send_header('Content-Type',mime)
self.send_header('Content-Length',str(len(data)));self.end_headers();self.wfile.write(data)
def do_POST(self):
body=self.rfile.read(int(self.headers['Content-Length']))
self.send_response(200);self.send_header('Content-Length',str(len(body)))
self.end_headers();self.wfile.write(body)
def log_message(self,*args):pass
class GatewayTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.backend=ThreadingHTTPServer(('127.0.0.1',0),Backend)
gateway.UPSTREAM_PORT=cls.backend.server_port
cls.proxy=ThreadingHTTPServer(('127.0.0.1',0),gateway.Gateway)
for s in (cls.backend,cls.proxy):threading.Thread(target=s.serve_forever,daemon=True).start()
@classmethod
def tearDownClass(cls):
for s in (cls.proxy,cls.backend):s.shutdown();s.server_close()
def request(self,path,method='GET',body=None):
c=http.client.HTTPConnection('127.0.0.1',self.proxy.server_port)
c.request(method,path,body);r=c.getresponse();result=(r.status,dict(r.getheaders()),r.read());c.close();return result
def test_mount_redirect_preserves_job(self):
code,headers,_=self.request('/box-sample?job=abc')
self.assertEqual(code,308);self.assertEqual(headers['Location'],'/box-sample/?job=abc')
def test_api_and_javascript_prefix_without_model_path_change(self):
_,headers,body=self.request('/box-sample/api/status')
self.assertIn(b'/box-sample/api/uploads/file',body);self.assertIn(b'/tmp/model.json',body)
self.assertEqual(int(headers['Content-Length']),len(body))
_,_,script=self.request('/box-sample/src/app.js')
self.assertIn(b"fetch('/box-sample/api/status')",script)
self.assertIn(b'/box-sample/dxf_rasterizer.html',script)
def test_binary_and_post_untouched(self):
self.assertEqual(self.request('/box-sample/api/uploads/file')[2],b'\x00\xff\x01')
self.assertEqual(self.request('/box-sample/api/analyze','POST',b'{"drawing":"/api/original"}')[2],b'{"drawing":"/api/original"}')
def test_other_service_paths_rejected(self):
for path in ('/api/status','/box-sample-other/api/status'):
self.assertEqual(self.request(path)[0],404)
if __name__=='__main__':unittest.main()