Skip to content

Fix Testing SOP Analyzer Script Functionality #1 #3

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
CodeeCod merged 2 commits into from
Nov 17, 2025
Merged

Fix Testing SOP Analyzer Script Functionality #1 #3

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ddb2e2b
Fix Testing SOP Analyzer Script Functionality #1
CodeeCod Nov 17, 2025
3c56fd5
Fix Testing SOP Analyzer Script Functionality #1
CodeeCod Nov 17, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
180 changes: 88 additions & 92 deletions test_sop_analyzer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,68 +1,73 @@
# test_sop_analyzer.py
import pytest
import json
import zipfile
import zlib
import tempfile
import os
import subprocess
import sys
from pathlib import Path
from sop_analyzer import SOPAnalyzer, RecordStats, TableInfo, OutputFormatter


class TestSOPAnalyzer:
"""Тесты для SOPAnalyzer"""

@pytest.fixture
def sample_data(self):
"""Создание тестовых данных"""
return {
"name": "test_package",
"pack_application_id": "test_app_123",
"timestamp": "2024-01-15T10:30:00Z",
"version": "1.0",
"records": [
{
"table_name": "users",
"action": "insert",
"is_strong_overwrite": False
},
{
"table_name": "users",
"action": "delete",
"is_strong_overwrite": False
},
{
"table_name": "products",
"action": "update",
"is_strong_overwrite": True
},
{
"table_name": "orders",
"action": "insert",
"is_strong_overwrite": False
}
]
}

@pytest.fixture
def create_test_sop_file(self, sample_data):
"""Создание тестового SOP файла"""
def _create_sop_file(compression_method=zlib.Z_BEST_COMPRESSION):
# Создаем временный файл
with tempfile.NamedTemporaryFile(suffix='.sop', delete=False) as f:
sop_path = f.name

# Создаем ZIP архив с данными
with zipfile.ZipFile(sop_path, 'w', compression=zipfile.ZIP_DEFLATED) as sop_zip:
# Конвертируем данные в JSON и сжимаем
json_data = json.dumps(sample_data).encode('utf-8')
compressed_data = zlib.compress(json_data, compression_method)

# Добавляем сжатые данные в архив
sop_zip.writestr('package.data', compressed_data)
@pytest.fixture
def sample_data():
"""Создание тестовых данных"""
return {
"name": "test_package",
"pack_application_id": "test_app_123",
"timestamp": "2024-01-15T10:30:00Z",
"version": "1.0",
"records": [
{
"table_name": "users",
"action": "insert",
"is_strong_overwrite": False
},
{
"table_name": "users",
"action": "delete",
"is_strong_overwrite": False
},
{
"table_name": "products",
"action": "update",
"is_strong_overwrite": True
},
{
"table_name": "orders",
"action": "insert",
"is_strong_overwrite": False
}
]
}


@pytest.fixture
def create_test_sop_file(sample_data):
"""Создание тестового SOP файла"""
def _create_sop_file(compression_method=zlib.Z_BEST_COMPRESSION):
# Создаем временный файл
with tempfile.NamedTemporaryFile(suffix='.sop', delete=False) as f:
sop_path = f.name

# Создаем ZIP архив с данными
with zipfile.ZipFile(sop_path, 'w', compression=zipfile.ZIP_DEFLATED) as sop_zip:
# Конвертируем данные в JSON и сжимаем
json_data = json.dumps(sample_data).encode('utf-8')
compressed_data = zlib.compress(json_data, compression_method)

return sop_path
# Добавляем сжатые данные в архив
sop_zip.writestr('package.data', compressed_data)

return _create_sop_file
return sop_path

return _create_sop_file


class TestSOPAnalyzer:
"""Тесты для SOPAnalyzer"""

def test_analyzer_initialization(self):
"""Тест инициализации анализатора"""
Expand Down Expand Up @@ -239,6 +244,32 @@ def test_empty_records(self):

finally:
os.unlink(sop_path)

def test_integration_with_cli(self, create_test_sop_file):
"""Интеграционный тест с CLI аргументами"""
sop_file = create_test_sop_file()
try:
# Тест базового вызова
result = subprocess.run([
sys.executable, 'sop_analyzer.py', sop_file
], capture_output=True, text=True, timeout=30)

assert result.returncode == 0
assert "МЕТАДАННЫЕ ПАКЕТА" in result.stdout
assert "СТАТИСТИКА ЗАПИСЕЙ" in result.stdout

# Тест JSON вывода
result = subprocess.run([
sys.executable, 'sop_analyzer.py', sop_file, '--json'
], capture_output=True, text=True, timeout=30)

assert result.returncode == 0
json_output = json.loads(result.stdout)
assert "metadata" in json_output
assert "record_statistics" in json_output

finally:
os.unlink(sop_file)


class TestOutputFormatter:
Expand Down Expand Up @@ -331,13 +362,8 @@ def test_print_tables_summary(self, capsys):
class TestDataCompression:
"""Тесты различных методов сжатия"""

def test_different_compression_levels(self):
def test_different_compression_levels(self, sample_data):
"""Тест различных уровней сжатия"""
test_data = {
"name": "compression_test",
"records": [{"table_name": "test", "action": "insert"}] * 10
}

compression_levels = [
zlib.Z_NO_COMPRESSION,
zlib.Z_BEST_SPEED,
Expand All @@ -352,50 +378,20 @@ def test_different_compression_levels(self):
try:
# Создаем SOP файл с указанным уровнем сжатия
with zipfile.ZipFile(sop_path, 'w') as sop_zip:
json_data = json.dumps(test_data).encode('utf-8')
json_data = json.dumps(sample_data).encode('utf-8')
compressed_data = zlib.compress(json_data, level)
sop_zip.writestr('package.data', compressed_data)

# Проверяем, что данные можно прочитать
analyzer = SOPAnalyzer(sop_path)
data = analyzer.load_data()

assert data["name"] == "compression_test"
assert len(data["records"]) == 10
assert data["name"] == "test_package"
assert len(data["records"]) == 4

finally:
os.unlink(sop_path)


def test_integration_with_cli(create_test_sop_file):
"""Интеграционный тест с CLI аргументами"""
import subprocess
import sys

sop_file = create_test_sop_file()
try:
# Тест базового вызова
result = subprocess.run([
sys.executable, 'sop_analyzer.py', sop_file
], capture_output=True, text=True, timeout=30)

assert result.returncode == 0
assert "МЕТАДАННЫЕ ПАКЕТА" in result.stdout
assert "СТАТИСТИКА ЗАПИСЕЙ" in result.stdout

# Тест JSON вывода
result = subprocess.run([
sys.executable, 'sop_analyzer.py', sop_file, '--json'
], capture_output=True, text=True, timeout=30)

assert result.returncode == 0
json_output = json.loads(result.stdout)
assert "metadata" in json_output
assert "record_statistics" in json_output

finally:
os.unlink(sop_file)


if __name__ == "__main__":
pytest.main([__file__, "-v"])
Loading