85 lines
2.7 KiB
Python
85 lines
2.7 KiB
Python
import asyncio
|
|
import os
|
|
from datetime import datetime
|
|
from dotenv import load_dotenv
|
|
from motor.motor_asyncio import AsyncIOMotorClient
|
|
|
|
from models.Asset import Asset, AssetType
|
|
from repos.assets_repo import AssetsRepo
|
|
from adapters.s3_adapter import S3Adapter
|
|
|
|
# Load env to get credentials
|
|
load_dotenv()
|
|
|
|
async def test_integration():
|
|
print("🚀 Starting integration test...")
|
|
|
|
# 1. Setup Dependencies
|
|
mongo_uri = os.getenv("MONGO_HOST", "mongodb://localhost:27017")
|
|
client = AsyncIOMotorClient(mongo_uri)
|
|
db_name = os.getenv("DB_NAME", "bot_db_test")
|
|
|
|
s3_adapter = S3Adapter(
|
|
endpoint_url=os.getenv("MINIO_ENDPOINT", "http://localhost:9000"),
|
|
aws_access_key_id=os.getenv("MINIO_ACCESS_KEY", "admin"),
|
|
aws_secret_access_key=os.getenv("MINIO_SECRET_KEY", "SuperSecretPassword123!"),
|
|
bucket_name=os.getenv("MINIO_BUCKET", "ai-char")
|
|
)
|
|
|
|
repo = AssetsRepo(client, s3_adapter, db_name=db_name)
|
|
|
|
# 2. Create Asset with Data and Thumbnail
|
|
print("📝 Creating asset...")
|
|
dummy_data = b"image_data_bytes"
|
|
dummy_thumb = b"thumbnail_bytes"
|
|
|
|
asset = Asset(
|
|
name="test_asset_with_thumb.png",
|
|
type=AssetType.IMAGE,
|
|
data=dummy_data,
|
|
thumbnail=dummy_thumb
|
|
)
|
|
|
|
asset_id = await repo.create_asset(asset)
|
|
print(f"✅ Asset created with ID: {asset_id}")
|
|
|
|
# 3. Verify object names in Mongo (Raw check)
|
|
print("🔍 Verifying Mongo metadata...")
|
|
# Used repo to fetch is better
|
|
fetched_asset = await repo.get_asset(asset_id, with_data=False)
|
|
|
|
if fetched_asset.minio_object_name:
|
|
print(f"✅ minio_object_name set: {fetched_asset.minio_object_name}")
|
|
else:
|
|
print("❌ minio_object_name NOT set!")
|
|
|
|
if fetched_asset.minio_thumbnail_object_name:
|
|
print(f"✅ minio_thumbnail_object_name set: {fetched_asset.minio_thumbnail_object_name}")
|
|
else:
|
|
print("❌ minio_thumbnail_object_name NOT set!")
|
|
|
|
# 4. Fetch Data from S3 via Repo
|
|
print("📥 Fetching data from MinIO...")
|
|
full_asset = await repo.get_asset(asset_id, with_data=True)
|
|
|
|
if full_asset.data == dummy_data:
|
|
print("✅ Data matches!")
|
|
else:
|
|
print(f"❌ Data mismatch! Got: {full_asset.data}")
|
|
|
|
if full_asset.thumbnail == dummy_thumb:
|
|
print("✅ Thumbnail matches!")
|
|
else:
|
|
print(f"❌ Thumbnail mismatch! Got: {full_asset.thumbnail}")
|
|
|
|
# 5. Clean up
|
|
print("🧹 Cleaning up...")
|
|
deleted = await repo.delete_asset(asset_id)
|
|
if deleted:
|
|
print("✅ Asset deleted")
|
|
else:
|
|
print("❌ Failed to delete asset")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(test_integration())
|