Refa: Move HTTP API tests to top-level test directory (#8042)

### What problem does this PR solve?

Move test cases only - CI still runs tests under sdk/python

### Type of change

- [x] Refactoring
This commit is contained in:
Liu An
2025-06-04 13:16:17 +08:00
committed by GitHub
parent b832372c98
commit 52c814b89d
39 changed files with 7934 additions and 6 deletions

63
test/utils/__init__.py Normal file
View File

@ -0,0 +1,63 @@
#
# Copyright 2025 The InfiniFlow Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import base64
import functools
import hashlib
import time
from pathlib import Path
def encode_avatar(image_path):
with Path.open(image_path, "rb") as file:
binary_data = file.read()
base64_encoded = base64.b64encode(binary_data).decode("utf-8")
return base64_encoded
def compare_by_hash(file1, file2, algorithm="sha256"):
def _calc_hash(file_path):
hash_func = hashlib.new(algorithm)
with open(file_path, "rb") as f:
while chunk := f.read(8192):
hash_func.update(chunk)
return hash_func.hexdigest()
return _calc_hash(file1) == _calc_hash(file2)
def wait_for(timeout=10, interval=1, error_msg="Timeout"):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
while True:
result = func(*args, **kwargs)
if result is True:
return result
elapsed = time.time() - start_time
if elapsed > timeout:
assert False, error_msg
time.sleep(interval)
return wrapper
return decorator
def is_sorted(data, field, descending=True):
timestamps = [ds[field] for ds in data]
return all(a >= b for a, b in zip(timestamps, timestamps[1:])) if descending else all(a <= b for a, b in zip(timestamps, timestamps[1:]))

107
test/utils/file_utils.py Normal file
View File

@ -0,0 +1,107 @@
#
# Copyright 2025 The InfiniFlow Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import json
from docx import Document # pip install python-docx
from openpyxl import Workbook # pip install openpyxl
from PIL import Image, ImageDraw # pip install Pillow
from pptx import Presentation # pip install python-pptx
from reportlab.pdfgen import canvas # pip install reportlab
def create_docx_file(path):
doc = Document()
doc.add_paragraph("这是一个测试 DOCX 文件。")
doc.save(path)
return path
def create_excel_file(path):
wb = Workbook()
ws = wb.active
ws["A1"] = "测试 Excel 文件"
wb.save(path)
return path
def create_ppt_file(path):
prs = Presentation()
slide = prs.slides.add_slide(prs.slide_layouts[0])
slide.shapes.title.text = "测试 PPT 文件"
prs.save(path)
return path
def create_image_file(path):
img = Image.new("RGB", (100, 100), color="blue")
draw = ImageDraw.Draw(img)
draw.text((10, 40), "Test", fill="white")
img.save(path)
return path
def create_pdf_file(path):
if not isinstance(path, str):
path = str(path)
c = canvas.Canvas(path)
c.drawString(100, 750, "测试 PDF 文件")
c.save()
return path
def create_txt_file(path):
with open(path, "w", encoding="utf-8") as f:
f.write("这是测试 TXT 文件的内容。")
return path
def create_md_file(path):
md_content = "# 测试 MD 文件\n\n这是一份 Markdown 格式的测试文件。"
with open(path, "w", encoding="utf-8") as f:
f.write(md_content)
return path
def create_json_file(path):
data = {"message": "这是测试 JSON 文件", "value": 123}
with open(path, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2)
return path
def create_eml_file(path):
eml_content = (
"From: sender@example.com\n"
"To: receiver@example.com\n"
"Subject: 测试 EML 文件\n\n"
"这是一封测试邮件的内容。\n"
)
with open(path, "w", encoding="utf-8") as f:
f.write(eml_content)
return path
def create_html_file(path):
html_content = (
"<html>\n"
"<head><title>测试 HTML 文件</title></head>\n"
"<body><h1>这是一个测试 HTML 文件</h1></body>\n"
"</html>"
)
with open(path, "w", encoding="utf-8") as f:
f.write(html_content)
return path

View File

@ -0,0 +1,28 @@
#
# Copyright 2025 The InfiniFlow Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import hypothesis.strategies as st
@st.composite
def valid_names(draw):
base_chars = "abcdefghijklmnopqrstuvwxyz_"
first_char = draw(st.sampled_from([c for c in base_chars if c.isalpha() or c == "_"]))
remaining = draw(st.text(alphabet=st.sampled_from(base_chars), min_size=0, max_size=128 - 2))
name = (first_char + remaining)[:128]
return name.encode("utf-8").decode("utf-8")