78 lines
2.3 KiB
Python
78 lines
2.3 KiB
Python
|
from fastapi.testclient import TestClient
|
||
|
|
import sys
|
||
|
|
from pathlib import Path as _Path
|
||
|
|
base = _Path(__file__).resolve().parents[2]
|
||
|
|
sys.path.insert(0, str(base))
|
||
|
|
sys.path.insert(0, str(base / "docling"))
|
||
|
|
import app.server as server
|
||
|
|
|
||
|
|
|
||
|
|
class _Resp:
|
||
|
|
def __init__(self, data: bytes):
|
||
|
|
self._data = data
|
||
|
|
def read(self) -> bytes:
|
||
|
|
return self._data
|
||
|
|
def close(self):
|
||
|
|
pass
|
||
|
|
|
||
|
|
|
||
|
|
class FakeMinio:
|
||
|
|
def __init__(self):
|
||
|
|
self.store = {
|
||
|
|
("doctest", "assets/rewritten/x.md"): (b"# Title\n\nhello", "text/markdown; charset=utf-8")
|
||
|
|
}
|
||
|
|
def stat_object(self, bucket: str, object_name: str):
|
||
|
|
class S:
|
||
|
|
def __init__(self, ct: str):
|
||
|
|
self.content_type = ct
|
||
|
|
k = (bucket, object_name)
|
||
|
|
if k in self.store:
|
||
|
|
return S(self.store[k][1])
|
||
|
|
return S("application/octet-stream")
|
||
|
|
def get_object(self, bucket: str, object_name: str):
|
||
|
|
k = (bucket, object_name)
|
||
|
|
if k in self.store:
|
||
|
|
return _Resp(self.store[k][0])
|
||
|
|
return _Resp(b"")
|
||
|
|
|
||
|
|
|
||
|
|
def setup():
|
||
|
|
server.RUNTIME_CONFIG["minio"].update({
|
||
|
|
"endpoint": "127.0.0.1:9000",
|
||
|
|
"public": "http://127.0.0.1:9000",
|
||
|
|
"access": "ak",
|
||
|
|
"secret": "sk",
|
||
|
|
"bucket": "doctest",
|
||
|
|
"secure": "false",
|
||
|
|
"prefix": "assets",
|
||
|
|
"store_final": "true",
|
||
|
|
"public_read": "true",
|
||
|
|
})
|
||
|
|
fake = FakeMinio()
|
||
|
|
def _cur():
|
||
|
|
return fake, "doctest", "http://127.0.0.1:9000", "assets"
|
||
|
|
server._minio_current = _cur # type: ignore
|
||
|
|
|
||
|
|
|
||
|
|
def run():
|
||
|
|
setup()
|
||
|
|
app = server.app
|
||
|
|
c = TestClient(app)
|
||
|
|
r = c.get("/minio/object", params={"bucket": "doctest", "object": "assets/rewritten/x.md"})
|
||
|
|
print("status:", r.status_code)
|
||
|
|
print("ct:", r.headers.get("Content-Type"))
|
||
|
|
print(r.text)
|
||
|
|
|
||
|
|
import urllib.parse as _u
|
||
|
|
enc = _u.quote("assets/rewritten/数字+产品手册-MD源文件/x.md")
|
||
|
|
cur_client, _, _, _ = server._minio_current() # type: ignore
|
||
|
|
cur_client.store[("doctest", "assets/rewritten/数字+产品手册-MD源文件/x.md")] = ("hello 中文+plus".encode("utf-8"), "text/markdown; charset=utf-8")
|
||
|
|
r2 = c.get("/minio/object", params={"bucket": "doctest", "object": enc})
|
||
|
|
print("status2:", r2.status_code)
|
||
|
|
print("ct2:", r2.headers.get("Content-Type"))
|
||
|
|
print(r2.text)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
run()
|