394 lines
15 KiB
Python
394 lines
15 KiB
Python
"""
|
||
Тесты для всех парсеров и их геттеров
|
||
"""
|
||
import pytest
|
||
import pandas as pd
|
||
import tempfile
|
||
import os
|
||
from unittest.mock import Mock, patch
|
||
import sys
|
||
|
||
# Добавляем путь к проекту
|
||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'python_parser'))
|
||
|
||
from adapters.parsers import (
|
||
SvodkaPMParser,
|
||
SvodkaCAParser,
|
||
MonitoringFuelParser,
|
||
MonitoringTarParser,
|
||
SvodkaRepairCAParser,
|
||
StatusesRepairCAParser,
|
||
OperSpravkaTechPosParser
|
||
)
|
||
|
||
|
||
class TestSvodkaPMParser:
|
||
"""Тесты для парсера Сводки ПМ"""
|
||
|
||
def setup_method(self):
|
||
"""Настройка перед каждым тестом"""
|
||
self.parser = SvodkaPMParser()
|
||
# Создаем тестовые данные
|
||
self.test_data = {
|
||
'SNPZ': pd.DataFrame({
|
||
'Процесс': ['Первичная переработка', 'Гидроочистка топлив'],
|
||
'Установка': ['SNPZ.EAVT6', 'SNPZ.L24-6'],
|
||
'План, т': [100.0, 200.0],
|
||
'Факт, т': [95.0, 190.0]
|
||
})
|
||
}
|
||
self.parser.data_dict = self.test_data
|
||
|
||
def test_parser_initialization(self):
|
||
"""Тест инициализации парсера"""
|
||
assert self.parser.name == "Сводки ПМ"
|
||
assert hasattr(self.parser, 'getters')
|
||
assert len(self.parser.getters) == 2
|
||
assert 'single_og' in self.parser.getters
|
||
assert 'total_ogs' in self.parser.getters
|
||
|
||
def test_single_og_getter(self):
|
||
"""Тест геттера single_og"""
|
||
params = {
|
||
'id': 'SNPZ',
|
||
'codes': [78, 79],
|
||
'columns': ['ПП', 'СЭБ']
|
||
}
|
||
|
||
result = self.parser.get_value('single_og', params)
|
||
|
||
assert result is not None
|
||
assert isinstance(result, str) # Возвращает JSON строку
|
||
|
||
def test_total_ogs_getter(self):
|
||
"""Тест геттера total_ogs"""
|
||
params = {
|
||
'codes': [78, 79],
|
||
'columns': ['ПП', 'СЭБ']
|
||
}
|
||
|
||
result = self.parser.get_value('total_ogs', params)
|
||
|
||
assert result is not None
|
||
assert isinstance(result, str) # Возвращает JSON строку
|
||
|
||
def test_getter_with_invalid_params(self):
|
||
"""Тест геттера с неверными параметрами"""
|
||
with pytest.raises(ValueError):
|
||
self.parser.get_value('single_og', {'invalid': 'params'})
|
||
|
||
def test_getter_with_nonexistent_og(self):
|
||
"""Тест геттера с несуществующим ОГ"""
|
||
params = {
|
||
'id': 'NONEXISTENT',
|
||
'codes': [78, 79],
|
||
'columns': ['ПП', 'СЭБ']
|
||
}
|
||
|
||
result = self.parser.get_value('single_og', params)
|
||
# Должен вернуть пустой результат, но не упасть
|
||
assert result is not None
|
||
|
||
|
||
class TestSvodkaCAParser:
|
||
"""Тесты для парсера Сводки СА"""
|
||
|
||
def setup_method(self):
|
||
"""Настройка перед каждым тестом"""
|
||
self.parser = SvodkaCAParser()
|
||
# Создаем тестовые данные
|
||
self.test_data = {
|
||
'plan': {
|
||
'ТиП': pd.DataFrame({
|
||
'ОГ': ['SNPZ', 'KNPZ'],
|
||
'Значение': [100.0, 200.0]
|
||
})
|
||
},
|
||
'fact': {
|
||
'ТиП': pd.DataFrame({
|
||
'ОГ': ['SNPZ', 'KNPZ'],
|
||
'Значение': [95.0, 190.0]
|
||
})
|
||
}
|
||
}
|
||
self.parser.data_dict = self.test_data
|
||
|
||
def test_parser_initialization(self):
|
||
"""Тест инициализации парсера"""
|
||
assert self.parser.name == "Сводки СА"
|
||
assert hasattr(self.parser, 'getters')
|
||
assert len(self.parser.getters) == 1
|
||
assert 'get_ca_data' in self.parser.getters
|
||
|
||
def test_get_ca_data_getter(self):
|
||
"""Тест геттера get_ca_data"""
|
||
params = {
|
||
'modes': ['plan', 'fact'],
|
||
'tables': ['ТиП', 'Топливо']
|
||
}
|
||
|
||
result = self.parser.get_value('get_ca_data', params)
|
||
|
||
assert result is not None
|
||
assert isinstance(result, dict) # Возвращает словарь
|
||
|
||
|
||
class TestMonitoringFuelParser:
|
||
"""Тесты для парсера Мониторинга топлива"""
|
||
|
||
def setup_method(self):
|
||
"""Настройка перед каждым тестом"""
|
||
self.parser = MonitoringFuelParser()
|
||
# Создаем тестовые данные
|
||
self.test_data = {
|
||
'SNPZ': pd.DataFrame({
|
||
'Дата': ['2024-01-01', '2024-01-02'],
|
||
'Топливо': ['Дизель', 'Бензин'],
|
||
'Количество': [100.0, 200.0],
|
||
'Объем': [50.0, 75.0] # Добавляем числовую колонку для агрегации
|
||
})
|
||
}
|
||
self.parser.data_dict = self.test_data
|
||
|
||
def test_parser_initialization(self):
|
||
"""Тест инициализации парсера"""
|
||
assert self.parser.name == "Мониторинг топлива"
|
||
assert hasattr(self.parser, 'getters')
|
||
# Проверяем, что есть геттеры
|
||
assert len(self.parser.getters) > 0
|
||
|
||
def test_getters_exist(self):
|
||
"""Тест существования геттеров"""
|
||
# Проверяем основные геттеры
|
||
getter_names = list(self.parser.getters.keys())
|
||
assert len(getter_names) > 0
|
||
|
||
# Тестируем каждый геттер с правильными параметрами
|
||
for getter_name in getter_names:
|
||
if getter_name == 'total_by_columns':
|
||
params = {'columns': ['Количество', 'Объем']} # Используем числовые колонки
|
||
else:
|
||
params = {}
|
||
|
||
try:
|
||
result = self.parser.get_value(getter_name, params)
|
||
assert result is not None
|
||
except ValueError as e:
|
||
# Некоторые геттеры могут требовать специфические параметры или иметь другие ошибки
|
||
error_msg = str(e).lower()
|
||
assert any(keyword in error_msg for keyword in ["required", "missing", "отсутствуют", "обязательные", "ошибка выполнения"])
|
||
|
||
|
||
class TestMonitoringTarParser:
|
||
"""Тесты для парсера Мониторинга ТЭР"""
|
||
|
||
def setup_method(self):
|
||
"""Настройка перед каждым тестом"""
|
||
self.parser = MonitoringTarParser()
|
||
# Создаем тестовые данные
|
||
self.test_data = {
|
||
'total': [pd.DataFrame({
|
||
'Дата': ['2024-01-01'],
|
||
'Потребление': [100.0]
|
||
})],
|
||
'last_day': [pd.DataFrame({
|
||
'Дата': ['2024-01-02'],
|
||
'Потребление': [150.0]
|
||
})]
|
||
}
|
||
self.parser.data_dict = self.test_data
|
||
|
||
def test_parser_initialization(self):
|
||
"""Тест инициализации парсера"""
|
||
assert self.parser.name == "monitoring_tar"
|
||
assert hasattr(self.parser, 'getters')
|
||
assert len(self.parser.getters) == 2
|
||
assert 'get_tar_data' in self.parser.getters
|
||
assert 'get_tar_full_data' in self.parser.getters
|
||
|
||
def test_get_tar_data_getter(self):
|
||
"""Тест геттера get_tar_data"""
|
||
params = {'mode': 'total'}
|
||
|
||
result = self.parser.get_value('get_tar_data', params)
|
||
|
||
assert result is not None
|
||
assert isinstance(result, str) # Возвращает JSON строку
|
||
|
||
def test_get_tar_full_data_getter(self):
|
||
"""Тест геттера get_tar_full_data"""
|
||
result = self.parser.get_value('get_tar_full_data', {})
|
||
|
||
assert result is not None
|
||
assert isinstance(result, str) # Возвращает JSON строку
|
||
|
||
|
||
class TestSvodkaRepairCAParser:
|
||
"""Тесты для парсера Сводки ремонта СА"""
|
||
|
||
def setup_method(self):
|
||
"""Настройка перед каждым тестом"""
|
||
self.parser = SvodkaRepairCAParser()
|
||
# Создаем тестовые данные в правильном формате
|
||
self.test_data = [
|
||
{
|
||
'id': 'SNPZ',
|
||
'Тип_ремонта': 'Капитальный',
|
||
'Статус': 'Завершен',
|
||
'Дата': '2024-01-01'
|
||
},
|
||
{
|
||
'id': 'SNPZ',
|
||
'Тип_ремонта': 'Текущий',
|
||
'Статус': 'В работе',
|
||
'Дата': '2024-01-02'
|
||
}
|
||
]
|
||
self.parser.data_dict = self.test_data
|
||
|
||
def test_parser_initialization(self):
|
||
"""Тест инициализации парсера"""
|
||
assert self.parser.name == "Сводки ремонта СА"
|
||
assert hasattr(self.parser, 'getters')
|
||
assert len(self.parser.getters) == 1
|
||
assert 'get_repair_data' in self.parser.getters
|
||
|
||
def test_get_repair_data_getter(self):
|
||
"""Тест геттера get_repair_data"""
|
||
params = {
|
||
'og_ids': ['SNPZ'],
|
||
'repair_types': ['КР'],
|
||
'include_planned': True,
|
||
'include_factual': True
|
||
}
|
||
|
||
result = self.parser.get_value('get_repair_data', params)
|
||
|
||
assert result is not None
|
||
assert isinstance(result, dict) # Возвращает словарь
|
||
|
||
|
||
class TestStatusesRepairCAParser:
|
||
"""Тесты для парсера Статусов ремонта СА"""
|
||
|
||
def setup_method(self):
|
||
"""Настройка перед каждым тестом"""
|
||
self.parser = StatusesRepairCAParser()
|
||
# Создаем тестовые данные
|
||
self.test_data = {
|
||
'SNPZ': pd.DataFrame({
|
||
'Статус': ['В работе', 'Завершен'],
|
||
'Процент': [50.0, 100.0]
|
||
})
|
||
}
|
||
self.parser.data_dict = self.test_data
|
||
|
||
def test_parser_initialization(self):
|
||
"""Тест инициализации парсера"""
|
||
assert self.parser.name == "Статусы ремонта СА"
|
||
assert hasattr(self.parser, 'getters')
|
||
assert len(self.parser.getters) == 1
|
||
assert 'get_repair_statuses' in self.parser.getters
|
||
|
||
def test_get_repair_statuses_getter(self):
|
||
"""Тест геттера get_repair_statuses"""
|
||
params = {'ids': ['SNPZ']}
|
||
|
||
result = self.parser.get_value('get_repair_statuses', params)
|
||
|
||
assert result is not None
|
||
assert isinstance(result, list) # Возвращает список
|
||
|
||
|
||
class TestOperSpravkaTechPosParser:
|
||
"""Тесты для парсера Операционных справок технологических позиций"""
|
||
|
||
def setup_method(self):
|
||
"""Настройка перед каждым тестом"""
|
||
self.parser = OperSpravkaTechPosParser()
|
||
# Создаем тестовые данные
|
||
self.test_data = {
|
||
'SNPZ': pd.DataFrame({
|
||
'Процесс': ['Первичная переработка', 'Гидроочистка топлив'],
|
||
'Установка': ['SNPZ.EAVT6', 'SNPZ.L24-6'],
|
||
'План, т': [100.0, 200.0],
|
||
'Факт, т': [95.0, 190.0],
|
||
'id': ['SNPZ.EAVT6', 'SNPZ.L24-6']
|
||
})
|
||
}
|
||
self.parser.data_dict = self.test_data
|
||
|
||
def test_parser_initialization(self):
|
||
"""Тест инициализации парсера"""
|
||
assert self.parser.name == "oper_spravka_tech_pos"
|
||
assert hasattr(self.parser, 'getters')
|
||
assert len(self.parser.getters) == 1
|
||
assert 'get_tech_pos' in self.parser.getters
|
||
|
||
def test_get_tech_pos_getter(self):
|
||
"""Тест геттера get_tech_pos"""
|
||
params = {'id': 'SNPZ'}
|
||
|
||
result = self.parser.get_value('get_tech_pos', params)
|
||
|
||
assert result is not None
|
||
assert isinstance(result, list) # Возвращает список словарей
|
||
|
||
def test_get_tech_pos_with_nonexistent_og(self):
|
||
"""Тест геттера с несуществующим ОГ"""
|
||
params = {'id': 'NONEXISTENT'}
|
||
|
||
result = self.parser.get_value('get_tech_pos', params)
|
||
|
||
assert result is not None
|
||
assert isinstance(result, list)
|
||
assert len(result) == 0 # Пустой список для несуществующего ОГ
|
||
|
||
|
||
class TestParserIntegration:
|
||
"""Интеграционные тесты для всех парсеров"""
|
||
|
||
def test_all_parsers_have_getters(self):
|
||
"""Тест, что все парсеры имеют геттеры"""
|
||
parsers = [
|
||
SvodkaPMParser(),
|
||
SvodkaCAParser(),
|
||
MonitoringFuelParser(),
|
||
MonitoringTarParser(),
|
||
SvodkaRepairCAParser(),
|
||
StatusesRepairCAParser(),
|
||
OperSpravkaTechPosParser()
|
||
]
|
||
|
||
for parser in parsers:
|
||
assert hasattr(parser, 'getters')
|
||
assert len(parser.getters) > 0
|
||
assert hasattr(parser, 'name')
|
||
assert parser.name is not None
|
||
|
||
def test_all_getters_return_valid_data(self):
|
||
"""Тест, что все геттеры возвращают валидные данные"""
|
||
parsers = [
|
||
SvodkaPMParser(),
|
||
SvodkaCAParser(),
|
||
MonitoringFuelParser(),
|
||
MonitoringTarParser(),
|
||
SvodkaRepairCAParser(),
|
||
StatusesRepairCAParser(),
|
||
OperSpravkaTechPosParser()
|
||
]
|
||
|
||
for parser in parsers:
|
||
for getter_name in parser.getters.keys():
|
||
try:
|
||
result = parser.get_value(getter_name, {})
|
||
assert result is not None
|
||
except Exception as e:
|
||
# Некоторые геттеры могут требовать специфические параметры
|
||
# Это нормально, главное что они не падают с критическими ошибками
|
||
error_msg = str(e).lower()
|
||
assert any(keyword in error_msg for keyword in ["required", "missing", "отсутствуют", "обязательные"])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
pytest.main([__file__]) |