Giới thiệu về việc tùy chỉnh Unittest

Unittest là framework kiểm thử tự động tích hợp sẵn trong Python, cung cấp các cấu trúc điều khiển và mô hình cơ bản. Tuy nhiên, trong các dự án thực tế, chức năng gốc thường chưa đủ đáp ứng nhu cầu phức tạp như:

• Tạo báo cáo kiểm thử dạng HTML.
• Thực thi song song nhiều luồng mà vẫn đảm bảo báo cáo rõ ràng.
• Tự động thực hiện lại các trường hợp lỗi (retry).
• Quản lý thẻ đánh dấu (tags) và cấp độ ưu tiên (level) cho từng case.

Để giải quyết các vấn đề này, việc phát triển thêm (secondary development) trên nền tảng Unittest là cần thiết. May mắn thay, Unittest cung cấp các API rõ ràng, giúp việc mở rộng và tùy biến trở nên dễ dàng.

Tổng quan về lớp unittest.TestResult

Lớp TestResult thường được khởi tạo bên trong TestRunner, đóng vai trò trung gian ghi nhận kết quả xuyên suốt quá trình thực thi các bộ kiểm thử (test suite) và từng trường hợp kiểm thử (test case). Các thuộc tính quan trọng bao gồm:

• stream: Luồng xuất dữ liệu (IO), thường là terminal hoặc file.
• verbosity: Mức độ chi tiết khi hiển thị thông tin.
• buffer: Nếu True, thông tin print sẽ được thu thập và xuất cùng lúc thay vì ngay lập tức.
• failfast: Dừng ngay lập tức khi gặp lỗi đầu tiên nếu được bật.

Các phương thức hook chính cho phép can thiệp vào quy trình:

• Vòng đời chạy: startTestRun, stopTestRun, startTest, stopTest.
• Ghi nhận kết quả: addSuccess, addFailure, addError, addSkip, v.v.
• Quản lý luồng xuất: _setupStdout, _restoreStdout.

Lưu ý: Failure (thất bại) thường do AssertionError (sai lệch kết quả mong đợi), trong khi Error (lỗi) là các ngoại lệ bất ngờ khác.

Mục tiêu tùy biến TestResult

Để phục vụ việc báo cáo và phân tích sâu hơn, lớp kết quả mới cần đạt được các yêu cầu sau:

1. Ghi nhận thời gian bắt đầu, kết thúc và thời lượng thực thi cho toàn bộ quá trình cũng như từng case.
2. Lưu trữ nội dung output (print) và thông tin exception để phục vụ báo cáo HTML.
3. Cung cấp lý do cụ thể cho các lỗi đã biết.
4. Dữ liệu tổng hợp (summary) và chi tiết phải có cấu trúc rõ ràng, dễ serialize (JSON).
5. Trích xuất mã nguồn của case kiểm thử để dễ dàng rà soát.
6. Thu thập thông tin môi trường thực thi (platform, python version).
7. Sử dụng logging thay vì print trực tiếp để dễ truy vết thời gian.
8. Hỗ trợ metadata như tags, level, mô tả cho từng case.

Thiết kế cấu trúc dữ liệu

Định dạng Summary

Dữ liệu tổng hợp cần bao gồm thống kê số lượng, thời gian chạy và thông tin môi trường. Cấu trúc đề xuất như sau:

{
    "name": "Tên bộ kết quả",
    "success": True/False,
    "stat": {
        "testsRun": 10,
        "successes": 8,
        "failures": 1,
        ...
    },
    "time": {
        "start_at": 1678888888.123,
        "end_at": 1678888890.456,
        "duration": 2.333
    },
    "platform": { ... },
    "details": [ ... ] // Danh sách chi tiết từng case
}

Định dạng chi tiết từng Case

Mỗi trường hợp kiểm thử cần lưu trữ thông tin tĩnh (trước khi chạy) và động (sau khi chạy):

{
    "name": "test_login",
    "id": "module.class.test_login",
    "tags": ["smoke", "auth"],
    "level": 1,
    "code": "def test_login...",
    "time": { "start_at": ..., "duration": ... },
    "status": "success/fail/error...",
    "output": "log nội bộ",
    "reason": "lý do lỗi/bỏ qua"
}

Triển khai chi tiết

Trích xuất Metadata từ Docstring

Để gắn thẻ (tags) và cấp độ (level) mà không cần sửa đổi code logic, ta có thể quy ước viết chúng trong docstring của phương thức kiểm thử. Sử dụng biểu thức chính quy để phân tích:

import re
import unittest

METADATA_TAG_REGEX = re.compile(r'tag:(\w+)')
METADATA_LEVEL_REGEX = re.compile(r'level:(\d+)')

def extract_tags_from_doc(test_case: unittest.TestCase) -> list:
    doc_content = getattr(test_case, '_testMethodDoc', None)
    tags = []
    if doc_content:
        found_tags = METADATA_TAG_REGEX.findall(doc_content)
        tags = list(set(found_tags))  # Loại bỏ trùng lặp
    return tags

def extract_level_from_doc(test_case: unittest.TestCase) -> int:
    doc_content = getattr(test_case, '_testMethodDoc', None)
    level = 0  # Mặc định
    if doc_content:
        found_levels = METADATA_LEVEL_REGEX.findall(doc_content)
        if found_levels:
            try:
                level = int(found_levels[0])
            except ValueError:
                pass
    return level

Lớp lưu trữ dữ liệu từng Case

Thay vì dùng dictionary thuần, ta xây dựng một lớp đối tượng để đóng gói dữ liệu của từng lần thực thi. Điều này giúp mã nguồn rõ ràng và dễ bảo trì hơn.

import inspect
import logging

logger = logging.getLogger(__name__)

class TestcaseDataRecord(object):
    def __init__(self, test_instance: unittest.TestCase):
        self.instance = test_instance
        self.name = test_instance._testMethodName
        self.case_id = test_instance.id()
        self.short_desc = test_instance.shortDescription()
        self.full_doc = test_instance._testMethodDoc
        
        self.module = test_instance.__module__
        self.class_name = test_instance.__class__.__name__
        self.class_path = f"{self.module}.{self.class_name}"
        
        # Trích xuất metadata
        self.tags = extract_tags_from_doc(test_instance)
        self.priority_level = extract_level_from_doc(test_instance)
        self.source_code = self._grab_source_code()
        
        # Thông tin động sẽ được cập nhật sau
        self.start_timestamp = None
        self.end_timestamp = None
        self.elapsed_time = None
        self.status = None
        self.captured_output = None
        self.traceback_info = None
        self.failure_reason = None

    def _grab_source_code(self):
        try:
            method_obj = getattr(self.instance.__class__, self.instance._testMethodName)
            return inspect.getsource(method_obj)
        except Exception:
            return ""

    @property
    def to_dict(self):
        return {
            "name": self.name,
            "id": self.case_id,
            "description": self.short_desc,
            "status": self.status,
            "metadata": {
                "tags": self.tags,
                "level": self.priority_level
            },
            "timing": {
                "start": self.start_timestamp,
                "end": self.end_timestamp,
                "duration": self.elapsed_time
            },
            "context": {
                "class": self.class_name,
                "module": self.module,
                "code": self.source_code
            },
            "result_data": {
                "output": self.captured_output,
                "error_log": self.traceback_info,
                "reason": self.failure_reason
            }
        }

Lớp TestResult mở rộng

Chúng ta sẽ kế thừa unittest.TestResult và ghi đè các phương thức cần thiết để tích hợp logging, đo thời gian và thu thập dữ liệu.

import unittest
import time
import sys
import io
import platform
from unittest.result import failfast

class EnrichedTestResult(unittest.TestResult):
    def __init__(self, stream=None, descriptions=None, verbosity=None):
        super().__init__(stream, descriptions, verbosity)
        self.verbosity = verbosity if verbosity is not None else 1
        self.buffer = True  # Bắt buộc buffer để capture output
        
        self.execution_records = []
        self.count_success = 0
        self.count_failure = 0
        self.count_error = 0
        self.count_skip = 0
        self.count_xfail = 0
        self.count_xpass = 0
        
        self.start_timestamp = None
        self.end_timestamp = None
        self.total_duration = None
        self.result_name = "CustomTestRun"
        
        # Mapping lỗi đã biết
        self.known_exceptions = {}

    def _get_platform_info(self):
        return {
            "system": platform.system(),
            "platform_detail": platform.platform(),
            "python_version": platform.python_version()
        }

    def startTestRun(self):
        self.start_timestamp = time.time()
        if self.verbosity > 1:
            logger.info(f"=== Bắt đầu bộ kiểm thử: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(self.start_timestamp))} ===")

    def stopTestRun(self):
        self.end_timestamp = time.time()
        self.total_duration = self.end_timestamp - self.start_timestamp
        if self.verbosity > 1:
            logger.info(f"=== Kết thúc bộ kiểm thử. Tổng thời gian: {self.total_duration:.2f}s ===")

    def startTest(self, test):
        super().startTest(test)
        record = TestcaseDataRecord(test)
        record.start_timestamp = time.time()
        test.current_record = record  # Gắn vào đối tượng test để các method khác truy cập
        self.execution_records.append(record)
        
        if self.verbosity > 1:
            logger.info(f"Đang chạy: {record.name} - {record.short_desc}")

    def _restoreStdout(self):
        if self.buffer:
            output = error = ''
            if self._mirrorOutput:
                output = sys.stdout.getvalue()
                error = sys.stderr.getvalue()
            
            sys.stdout = self._original_stdout
            sys.stderr = self._original_stderr
            
            if self._stdout_buffer:
                self._stdout_buffer.seek(0)
                self._stdout_buffer.truncate()
            if self._stderr_buffer:
                self._stderr_buffer.seek(0)
                self._stderr_buffer.truncate()
                
            return output + error if (output or error) else None
        return None

    def stopTest(self, test):
        record = getattr(test, 'current_record', None)
        if record:
            record.end_timestamp = time.time()
            record.elapsed_time = record.end_timestamp - record.start_timestamp
            record.captured_output = self._restoreStdout()
            
            if self.verbosity > 1 and record.status:
                logger.info(f"Kết quả: {record.status} - Thời gian: {record.elapsed_time:.3f}s")
            elif self.verbosity > 0:
                status_char = '.' if record.status == 'success' else 'F'
                print(f"{status_char}", end='', flush=True)

    def _extract_error_message(self, err):
        exctype, value, tb = err
        msg = str(value)
        full_path = f"{exctype.__module__}.{exctype.__name__}"
        if self.known_exceptions and full_path in self.known_exceptions:
            return self.known_exceptions[full_path]
        return msg

    def _format_traceback(self, err, test):
        exctype, value, tb = err
        # Logic loại bỏ các frame không cần thiết tương tự unittest gốc
        while tb and self._is_relevant_tb_level(tb):
            tb = tb.tb_next
        
        length = None
        if exctype is test.failureException:
            length = self._count_relevant_tb_levels(tb)
            
        tb_exception = traceback.TracebackException(
            exctype, value, tb, limit=length, capture_locals=self.tb_locals
        )
        return ''.join(tb_exception.format())

    def addSuccess(self, test):
        test.current_record.status = 'success'
        self.count_success += 1
        super().addSuccess(test)

    @failfast
    def addFailure(self, test, err):
        test.current_record.status = 'failure'
        test.current_record.traceback_info = self._format_traceback(err, test)
        test.current_record.failure_reason = self._extract_error_message(err)
        self.count_failure += 1
        super().addFailure(test, err)

    @failfast
    def addError(self, test, err):
        test.current_record.status = 'error'
        test.current_record.traceback_info = self._format_traceback(err, test)
        test.current_record.failure_reason = self._extract_error_message(err)
        self.count_error += 1
        super().addError(test, err)

    def addSkip(self, test, reason):
        test.current_record.status = 'skipped'
        test.current_record.failure_reason = reason
        self.count_skip += 1
        super().addSkip(test, reason)

    def addExpectedFailure(self, test, err):
        test.current_record.status = 'expected_failure'
        test.current_record.traceback_info = self._format_traceback(err, test)
        self.count_xfail += 1
        super().addExpectedFailure(test, err)

    def addUnexpectedSuccess(self, test):
        test.current_record.status = 'unexpected_success'
        self.count_xpass += 1
        super().addUnexpectedSuccess(test)

    @property
    def summary_report(self):
        return {
            "name": self.result_name,
            "success": self.wasSuccessful(),
            "statistics": {
                "total": self.testsRun,
                "success": self.count_success,
                "failure": self.count_failure,
                "error": self.count_error,
                "skipped": self.count_skip,
                "expected_failure": self.count_xfail,
                "unexpected_success": self.count_xpass
            },
            "timing": {
                "start": self.start_timestamp,
                "end": self.end_timestamp,
                "duration": self.total_duration
            },
            "environment": self._get_platform_info(),
            "details": [rec.to_dict for rec in self.execution_records]
        }

Ví dụ sử dụng

Dưới đây là cách tích hợp lớp EnrichedTestResult vào quy trình chạy kiểm thử thông thường. Lưu ý cấu hình logging để thay thế cho các lệnh print trực tiếp.

import unittest
import logging

# Cấu hình logging cơ bản
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

class SampleTestSuite(unittest.TestCase):
    def test_feature_a(self):
        """Kiểm tra tính năng A
        tag:smoke
        tag:core
        level:1
        """
        logging.info("Đang thực hiện kiểm tra A")
        self.assertTrue(True)

    def test_feature_b(self):
        """Kiểm tra tính năng B
        tag:regression
        level:2
        """
        logging.warning("Cảnh báo trong test B")
        self.assertEqual(1, 1)

if __name__ == '__main__':
    loader = unittest.defaultTestLoader
    suite = loader.loadTestsFromTestCase(SampleTestSuite)
    
    # Khởi tạo runner với lớp result tùy chỉnh
    runner = unittest.TextTestRunner(resultclass=EnrichedTestResult, verbosity=2)
    result = runner.run(suite)
    
    # In ra báo cáo tổng hợp dạng dictionary
    import pprint
    pprint.pprint(result.summary_report)

Kết quả thực thi sẽ hiển thị thông tin log thời gian thực và cuối cùng trả về một cấu trúc dữ liệu đầy đủ chứa thông tin môi trường, thời gian chạy, mã nguồn và kết quả chi tiết của từng case, sẵn sàng để được chuyển đổi sang định dạng HTML hoặc lưu trữ vào cơ sở dữ liệu.