105 lines
2.9 KiB
Python
105 lines
2.9 KiB
Python
"""Test MongoDB actions and models."""
|
|
|
|
import pytest
|
|
from pymongo import MongoClient
|
|
|
|
from Services.MongoDb.Models.actions import MongoActions
|
|
from Services.MongoDb.Models.action_models.domain import (
|
|
DomainData,
|
|
DomainDocumentCreate,
|
|
DomainDocumentUpdate,
|
|
)
|
|
from AllConfigs.NoSqlDatabase.configs import MongoConfig
|
|
|
|
|
|
@pytest.fixture
|
|
def mongo_client():
|
|
"""Create MongoDB test client."""
|
|
# Connect using configured credentials
|
|
client = MongoClient(MongoConfig.URL)
|
|
client.admin.command("ping") # Test connection
|
|
yield client
|
|
client.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def mongo_actions(mongo_client):
|
|
"""Create MongoActions instance for testing."""
|
|
if not mongo_client:
|
|
pytest.skip("MongoDB connection not available")
|
|
|
|
actions = MongoActions(
|
|
client=mongo_client,
|
|
database=MongoConfig.DATABASE_NAME,
|
|
company_uuid="test_company",
|
|
storage_reason="domains",
|
|
)
|
|
yield actions
|
|
try:
|
|
# Cleanup after tests
|
|
if actions.collection is not None:
|
|
actions.collection.drop()
|
|
except Exception as e:
|
|
print(f"Failed to cleanup test collection: {e}")
|
|
|
|
|
|
def test_mongo_crud_operations(mongo_actions: MongoActions):
|
|
"""Test CRUD operations with MongoActions."""
|
|
|
|
# Create test data
|
|
domain_data = DomainData(
|
|
user_uu_id="test_user",
|
|
main_domain="example.com",
|
|
other_domains_list=["old.com"],
|
|
)
|
|
create_doc = DomainDocumentCreate(data=domain_data)
|
|
|
|
# Test create
|
|
result = mongo_actions.insert_one(create_doc.model_dump())
|
|
assert result.inserted_id is not None
|
|
|
|
# Test read
|
|
doc = mongo_actions.find_one({"data.main_domain": "example.com"})
|
|
assert doc is not None
|
|
assert doc["data"]["main_domain"] == "example.com"
|
|
|
|
# Test update
|
|
update_data = DomainData(
|
|
user_uu_id="test_user",
|
|
main_domain="new.com",
|
|
other_domains_list=["example.com", "old.com"],
|
|
)
|
|
update_doc = DomainDocumentUpdate(data=update_data)
|
|
result = mongo_actions.update_one(
|
|
{"_id": doc["_id"]}, {"$set": update_doc.model_dump()}
|
|
)
|
|
assert result.modified_count == 1
|
|
|
|
# Test delete
|
|
result = mongo_actions.delete_one({"_id": doc["_id"]})
|
|
assert result.deleted_count == 1
|
|
|
|
|
|
def test_mongo_aggregate(mongo_actions: MongoActions):
|
|
"""Test aggregate operations with MongoActions."""
|
|
|
|
# Insert test documents
|
|
docs = [
|
|
DomainDocumentCreate(
|
|
data=DomainData(user_uu_id="user1", main_domain=f"domain{i}.com")
|
|
).model_dump()
|
|
for i in range(3)
|
|
]
|
|
mongo_actions.insert_many(docs)
|
|
|
|
# Test aggregation
|
|
pipeline = [{"$group": {"_id": "$data.user_uu_id", "count": {"$sum": 1}}}]
|
|
result = mongo_actions.aggregate(pipeline)
|
|
result_list = list(result)
|
|
assert len(result_list) == 1
|
|
assert result_list[0]["count"] == 3
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|