import sys
import os
import unittest
import asyncio
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from library.utils import old_add, old_multiply, square, cube, legacy_divide
from library.models import OldDataProcessor, LegacyUser, SimpleProcessor
from library.async_ops import old_fetch_data, legacy_upload
from library.processors import ProcessingEngine, DataValidator
from library.containers import LegacyContainer, SmartList
class TestDeprecatedFunctions(unittest.TestCase):
def test_deprecated_math_functions(self):
result = old_add(5, 3)
self.assertEqual(result, 8)
result = old_multiply(4, 6)
self.assertEqual(result, 24)
result = square(7)
self.assertEqual(result, 49)
result = cube(4)
self.assertEqual(result, 64)
result = legacy_divide(15, 3)
self.assertEqual(result, 5.0)
def test_deprecated_processor_class(self):
processor = OldDataProcessor("test_db", cache_size=10)
result = processor.process("test_data")
self.assertIn("test_data", result)
self.assertIn("batch_size=20", result)
info = processor.get_info()
self.assertEqual(info["data_source"], "test_db")
self.assertEqual(info["batch_size"], 20)
def test_legacy_user_methods(self):
user = LegacyUser("test_user", "test@example.com", active=True)
name = user.get_name()
self.assertEqual(name, "test_user")
profile = user.set_profile_data({"age": 25, "city": "NYC"})
self.assertEqual(profile["age"], 25)
self.assertEqual(profile["city"], "NYC")
enabled = user.is_enabled
self.assertTrue(enabled)
def test_simple_processor(self):
processor = SimpleProcessor("test_source")
result = processor.process_item("test_item")
self.assertIn("test_item", result)
self.assertIn("test_source", result)
def test_deprecated_static_methods(self):
is_valid = ProcessingEngine.old_validate_data(["test", "data"])
self.assertTrue(is_valid)
cleaned = ProcessingEngine.legacy_clean_text(" HELLO WORLD ")
self.assertEqual(cleaned, "hello world")
result = ProcessingEngine.old_process_batch(
["Item1", "Item2"],
uppercase=False,
trim=True
)
self.assertEqual(result, ["item1", "item2"])
def test_deprecated_class_methods(self):
engine = ProcessingEngine.old_create_engine(fast=True)
self.assertEqual(engine.engine_type, "fast")
engine = ProcessingEngine.old_create_engine(fast=False)
self.assertEqual(engine.engine_type, "memory_efficient")
fast_engine = ProcessingEngine.legacy_fast_processor()
self.assertEqual(fast_engine.engine_type, "fast")
def test_deprecated_validators(self):
valid = DataValidator.check_email("test@example.com")
self.assertTrue(valid)
invalid = DataValidator.check_email("invalid_email")
self.assertFalse(invalid)
valid_phone = DataValidator.validate_phone_number("555-123-4567")
self.assertTrue(valid_phone)
validator = DataValidator.create_strict_validator()
self.assertTrue(validator.strict_mode)
def test_deprecated_container_methods(self):
container = LegacyContainer(["a", "b", "c"])
size = container.size()
self.assertEqual(size, 3)
string_rep = container.to_str()
self.assertIn("3 items", string_rep)
item = container.get_item(0)
self.assertEqual(item, "a")
container.set_item(1, "modified")
self.assertEqual(container._items[1], "modified")
contains = container.contains("a")
self.assertTrue(contains)
def test_smart_list_deprecated_methods(self):
smart_list = SmartList([1, 2, 3])
length = smart_list.length()
self.assertEqual(length, 3)
count = smart_list.get_count()
self.assertEqual(count, 3)
empty = smart_list.is_empty()
self.assertFalse(empty)
empty_list = SmartList([])
empty_check = empty_list.is_empty()
self.assertTrue(empty_check)
has_item = smart_list.has_item(2)
self.assertTrue(has_item)
no_item = smart_list.has_item(99)
self.assertFalse(no_item)
class TestDeprecatedAsync(unittest.IsolatedAsyncioTestCase):
async def test_old_fetch_data(self):
result = await old_fetch_data("https://test.example.com")
self.assertEqual(result["url"], "https://test.example.com")
self.assertEqual(result["timeout"], 30)
self.assertEqual(result["status"], 200)
async def test_legacy_upload(self):
result = await legacy_upload("test.txt", "https://upload.example.com")
self.assertEqual(result["file_path"], "test.txt")
self.assertEqual(result["destination"], "https://upload.example.com")
self.assertEqual(result["chunk_size"], 4096)
self.assertEqual(result["status"], "uploaded")
class TestMigrationScenarios(unittest.TestCase):
def test_nested_deprecated_calls(self):
inner_result = old_add(5, 3)
outer_result = old_multiply(inner_result, 2)
final_result = square(outer_result)
self.assertEqual(final_result, 256)
def test_deprecated_in_comprehensions(self):
numbers = [1, 2, 3, 4, 5]
squares = [square(n) for n in numbers]
self.assertEqual(squares, [1, 4, 9, 16, 25])
cubes = [cube(n) for n in numbers if n > 2]
self.assertEqual(cubes, [27, 64, 125])
def test_deprecated_in_lambda(self):
numbers = [1, 2, 3, 4, 5]
add_ten = lambda x: old_add(x, 10)
results = list(map(add_ten, numbers))
self.assertEqual(results, [11, 12, 13, 14, 15])
even_squares = list(filter(lambda x: square(x) % 2 == 0, numbers))
self.assertEqual(even_squares, [2, 4])
def test_deprecated_with_error_handling(self):
try:
result = legacy_divide(10, 2)
self.assertEqual(result, 5.0)
except Exception as e:
self.fail(f"Unexpected exception: {e}")
with self.assertRaises(ValueError):
legacy_divide(10, 0)
def test_mixed_old_and_new_apis(self):
from library.utils import add, multiply
old_result = old_add(5, 3) new_result = add(7, 2) combined = multiply(old_result, new_result)
self.assertEqual(combined, 72)
def run_tests():
print("=== Running Integration Tests with Deprecated Functions ===")
print("Note: You will see deprecation warnings during test execution.")
print("This is expected and demonstrates the library's warning system.\n")
loader = unittest.TestLoader()
suite = unittest.TestSuite()
suite.addTests(loader.loadTestsFromTestCase(TestDeprecatedFunctions))
suite.addTests(loader.loadTestsFromTestCase(TestDeprecatedAsync))
suite.addTests(loader.loadTestsFromTestCase(TestMigrationScenarios))
runner = unittest.TextTestRunner(verbosity=2)
result = runner.run(suite)
print(f"\n=== Test Results ===")
print(f"Tests run: {result.testsRun}")
print(f"Failures: {len(result.failures)}")
print(f"Errors: {len(result.errors)}")
if result.failures:
print(f"\nFailures:")
for test, traceback in result.failures:
print(f" {test}: {traceback}")
if result.errors:
print(f"\nErrors:")
for test, traceback in result.errors:
print(f" {test}: {traceback}")
print(f"\n=== Migration Instructions ===")
print("To migrate this test file:")
print(" dissolve migrate consumer/test_integration.py")
print("To apply migrations:")
print(" dissolve migrate --write consumer/test_integration.py")
return result.wasSuccessful()
if __name__ == "__main__":
success = run_tests()
sys.exit(0 if success else 1)