test_data_diagnosis.py

# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.

"""Tests for DataDiagnosis module."""

import json
import unittest
import yaml
from pathlib import Path

import pandas as pd
import numpy as np

from superbench.analyzer import DataDiagnosis
import superbench.analyzer.file_handler as file_handler


class TestDataDiagnosis(unittest.TestCase):
    """Test for DataDiagnosis class."""
    def setUp(self):
        """Method called to prepare the test fixture."""
        self.parent_path = Path(__file__).parent
        self.output_excel_file = str(self.parent_path / 'diagnosis_summary.xlsx')
        self.test_rule_file_fake = str(self.parent_path / 'test_rules_fake.yaml')
        self.output_json_file = str(self.parent_path / 'diagnosis_summary.json')
        self.output_jsonl_file = str(self.parent_path / 'diagnosis_summary.jsonl')
        self.output_md_file = str(self.parent_path / 'diagnosis_summary.md')
        self.output_html_file = str(self.parent_path / 'diagnosis_summary.html')
        self.output_all_json_file = str(self.parent_path / 'diagnosis_summary.json')

    def tearDown(self):
        """Method called after the test method has been called and the result recorded."""
        for file in [
            self.output_excel_file, self.output_json_file, self.output_jsonl_file, self.test_rule_file_fake,
            self.output_md_file, self.output_html_file, self.output_all_json_file
        ]:
            p = Path(file)
            if p.is_file():
                p.unlink()

    def test_data_diagnosis(self):
        """Test for rule-based data diagnosis."""
        # Test - read_raw_data and get_metrics_from_raw_data
        # Positive case
        test_raw_data = str(self.parent_path / 'test_results.jsonl')
        test_rule_file = str(self.parent_path / 'test_rules.yaml')
        test_baseline_file = str(self.parent_path / 'test_baseline.json')
        diag1 = DataDiagnosis()
        diag1._raw_data_df = file_handler.read_raw_data(test_raw_data)
        diag1._benchmark_metrics_dict = diag1._get_metrics_by_benchmarks(list(diag1._raw_data_df))
        assert (len(diag1._raw_data_df) == 3)
        # Negative case
        test_raw_data_fake = str(self.parent_path / 'test_results_fake.jsonl')
        test_rule_file_fake = str(self.parent_path / 'test_rules_fake.yaml')
        diag2 = DataDiagnosis()
        self.assertRaises(FileNotFoundError, file_handler.read_raw_data, test_raw_data_fake)
        diag2._benchmark_metrics_dict = diag2._get_metrics_by_benchmarks([])
        assert (len(diag2._benchmark_metrics_dict) == 0)
        metric_list = [
            'gpu_temperature', 'gpu_power_limit', 'gemm-flops/FP64',
            'bert_models/pytorch-bert-base/steptime_train_float32'
        ]
        self.assertDictEqual(
            diag2._get_metrics_by_benchmarks(metric_list), {
                'gemm-flops': {'gemm-flops/FP64'},
                'bert_models': {'bert_models/pytorch-bert-base/steptime_train_float32'}
            }
        )
        # Test - read rules
        self.assertRaises(FileNotFoundError, file_handler.read_rules, test_rule_file_fake)
        rules = file_handler.read_rules(test_rule_file)
        assert (rules)
        # Test - _check_and_format_rules
        # Negative case
        false_rules = [
            {
                'criteria': 'lambda x:x>0',
                'categories': 'KernelLaunch',
                'metrics': ['kernel-launch/event_overhead:\\d+']
            }, {
                'criteria': 'lambda x:x>0',
                'function': 'variance',
                'metrics': ['kernel-launch/event_overhead:\\d+']
            }, {
                'categories': 'KernelLaunch',
                'function': 'variance',
                'metrics': ['kernel-launch/event_overhead:\\d+']
            }, {
                'criteria': 'lambda x:x>0',
                'function': 'abb',
                'categories': 'KernelLaunch',
                'metrics': ['kernel-launch/event_overhead:\\d+']
            }, {
                'criteria': 'lambda x:x>0',
                'function': 'abb',
                'categories': 'KernelLaunch',
            }, {
                'criteria': 'x>5',
                'function': 'abb',
                'categories': 'KernelLaunch',
                'metrics': ['kernel-launch/event_overhead:\\d+']
            }
        ]
        metric = 'kernel-launch/event_overhead:0'
        for rules in false_rules:
            self.assertRaises(Exception, diag1._check_and_format_rules, rules, metric)
        # Positive case
        true_rules = [
            {
                'categories': 'KernelLaunch',
                'criteria': 'lambda x:x>0.05',
                'function': 'variance',
                'metrics': ['kernel-launch/event_overhead:\\d+']
            }, {
                'categories': 'KernelLaunch',
                'criteria': 'lambda x:x<-0.05',
                'function': 'variance',
                'metrics': 'kernel-launch/event_overhead:\\d+'
            }, {
                'categories': 'KernelLaunch',
                'criteria': 'lambda x:x>0',
                'function': 'value',
                'metrics': ['kernel-launch/event_overhead:\\d+']
            }
        ]
        for rules in true_rules:
            assert (diag1._check_and_format_rules(rules, metric))
        # Test - _get_baseline_of_metric
        baseline = file_handler.read_baseline(test_baseline_file)
        assert (diag1._get_baseline_of_metric(baseline, 'kernel-launch/event_overhead:0') == 0.00596)
        assert (diag1._get_baseline_of_metric(baseline, 'kernel-launch/return_code') == 0)
        assert (diag1._get_baseline_of_metric(baseline, 'mem-bw/H2D:0') is None)
        # Test - _parse_rules_and_baseline
        # Negative case
        fake_rules = []
        baseline = file_handler.read_baseline(test_baseline_file)
        self.assertRaises(Exception, diag2._parse_rules_and_baseline, fake_rules, baseline)
        diag2 = DataDiagnosis()
        diag2._raw_data_df = file_handler.read_raw_data(test_raw_data)
        diag2._benchmark_metrics_dict = diag2._get_metrics_by_benchmarks(list(diag2._raw_data_df))
        p = Path(test_rule_file)
        with p.open() as f:
            rules = yaml.load(f, Loader=yaml.SafeLoader)
        rules['superbench']['rules']['fake'] = false_rules[0]
        with open(test_rule_file_fake, 'w') as f:
            yaml.dump(rules, f)
        self.assertRaises(Exception, diag1._parse_rules_and_baseline, fake_rules, baseline)
        # Positive case
        rules = file_handler.read_rules(test_rule_file)
        assert (diag1._parse_rules_and_baseline(rules, baseline))
        # Test - _run_diagnosis_rules_for_single_node
        (details_row, summary_data_row) = diag1._run_diagnosis_rules_for_single_node('sb-validation-01')
        assert (details_row)
        (details_row, summary_data_row) = diag1._run_diagnosis_rules_for_single_node('sb-validation-02')
        assert (not details_row)
        # Test - _run_diagnosis_rules
        baseline = file_handler.read_baseline(test_baseline_file)
        data_not_accept_df, label_df = diag1.run_diagnosis_rules(rules, baseline)
        assert (len(label_df) == 3)
        assert (label_df.loc['sb-validation-01']['label'] == 1)
        assert (label_df.loc['sb-validation-02']['label'] == 0)
        assert (label_df.loc['sb-validation-03']['label'] == 1)
        node = 'sb-validation-01'
        row = data_not_accept_df.loc[node]
        assert (len(row) == 36)
        assert (row['Category'] == 'KernelLaunch')
        assert (
            row['Defective Details'] ==
            'kernel-launch/event_overhead:0(B/L: 0.0060 VAL: 0.1000 VAR: 1577.85% Rule:lambda x:x>0.05)'
        )
        node = 'sb-validation-03'
        row = data_not_accept_df.loc[node]
        assert (len(row) == 36)
        assert ('FailedTest' in row['Category'])
        assert ('mem-bw/return_code(VAL: 1.0000 Rule:lambda x:x>0)' in row['Defective Details'])
        assert ('mem-bw/H2D_Mem_BW:0_miss' in row['Defective Details'])
        assert (len(data_not_accept_df) == 2)
        # Test - output in excel
        diag1.output_diagnosis_in_excel(diag1._raw_data_df, data_not_accept_df, self.output_excel_file, diag1._sb_rules)
        excel_file = pd.ExcelFile(self.output_excel_file, engine='openpyxl')
        data_sheet_name = 'Raw Data'
        raw_data_df = excel_file.parse(data_sheet_name)
        assert (len(raw_data_df) == 3)
        data_sheet_name = 'Not Accept'
        data_not_accept_read_from_excel = excel_file.parse(data_sheet_name)
        assert (len(data_not_accept_read_from_excel) == 2)
        assert ('Category' in data_not_accept_read_from_excel)
        assert ('Defective Details' in data_not_accept_read_from_excel)
        # Test - output in jsonl
        diag1.output_diagnosis_in_jsonl(data_not_accept_df, self.output_json_file)
        assert (Path(self.output_json_file).is_file())
        with Path(self.output_json_file).open() as f:
            data_not_accept_read_from_json = f.readlines()
        assert (len(data_not_accept_read_from_json) == 2)
        for line in data_not_accept_read_from_json:
            json.loads(line)
            assert ('Category' in line)
            assert ('Defective Details' in line)
            assert ('index' in line)
        # Test - generate_md_lines
        lines = diag1.generate_md_lines(data_not_accept_df, diag1._sb_rules, 2)
        assert (lines)
        expected_md_file = str(self.parent_path / '../data/diagnosis_summary.md')
        with open(expected_md_file, 'r') as f:
            expect_result = f.readlines()
        assert (lines == expect_result)
        # Test - output_all_nodes_results
        # case 1: 1 accept, 2 not accept
        data_df = diag1.output_all_nodes_results(diag1._raw_data_df, data_not_accept_df)
        assert (len(data_df) == 3)
        assert (not data_df.loc['sb-validation-01']['Accept'])
        assert (data_df.loc['sb-validation-02']['Accept'])
        assert (not data_df.loc['sb-validation-03']['Accept'])
        assert ('Category' in data_df)
        assert ('Defective Details' in data_df)
        # case 1: 3 accept, 0 not accept
        data_df_all_accept = diag1.output_all_nodes_results(diag1._raw_data_df, pd.DataFrame())
        assert (len(data_df_all_accept) == 3)
        assert (data_df_all_accept.loc['sb-validation-01']['Accept'])
        assert (data_df_all_accept.loc['sb-validation-02']['Accept'])
        assert (data_df_all_accept.loc['sb-validation-03']['Accept'])
        # Test - output in json
        diag1.output_diagnosis_in_json(data_df, self.output_all_json_file)
        assert (Path(self.output_all_json_file).is_file())
        expected_result_file = str(self.parent_path / '../data/diagnosis_summary.json')
        with Path(self.output_all_json_file).open() as f:
            data_not_accept_read_from_json = f.read()
        with Path(expected_result_file).open() as f:
            expect_result = f.read()
        assert (data_not_accept_read_from_json == expect_result)

    def test_data_diagnosis_run(self):
        """Test for the run process of rule-based data diagnosis."""
        test_raw_data = str(self.parent_path / 'test_results.jsonl')
        test_rule_file = str(self.parent_path / 'test_rules.yaml')
        test_baseline_file = str(self.parent_path / 'test_baseline.json')

        # Test - output in excel
        DataDiagnosis().run(test_raw_data, test_rule_file, test_baseline_file, str(self.parent_path), 'excel')
        excel_file = pd.ExcelFile(self.output_excel_file, engine='openpyxl')
        data_sheet_name = 'Not Accept'
        data_not_accept_read_from_excel = excel_file.parse(data_sheet_name)
        expect_result_file = pd.ExcelFile(str(self.parent_path / '../data/diagnosis_summary.xlsx'), engine='openpyxl')
        expect_result = expect_result_file.parse(data_sheet_name)
        pd.testing.assert_frame_equal(data_not_accept_read_from_excel, expect_result)
        # Test - output in json
        DataDiagnosis().run(test_raw_data, test_rule_file, test_baseline_file, str(self.parent_path), 'json')
        assert (Path(self.output_json_file).is_file())
        with Path(self.output_json_file).open() as f:
            data_not_accept_read_from_json = f.read()
        expect_result_file = self.parent_path / '../data/diagnosis_summary_json.json'
        with Path(expect_result_file).open() as f:
            expect_result = f.read()
        assert (data_not_accept_read_from_json == expect_result)
        # Test - output in jsonl
        DataDiagnosis().run(test_raw_data, test_rule_file, test_baseline_file, str(self.parent_path), 'jsonl')
        assert (Path(self.output_jsonl_file).is_file())
        with Path(self.output_jsonl_file).open() as f:
            data_not_accept_read_from_jsonl = f.read()
        expect_result_file = self.parent_path / '../data/diagnosis_summary.jsonl'
        with Path(expect_result_file).open() as f:
            expect_result = f.read()
        assert (data_not_accept_read_from_jsonl == expect_result)
        # Test - output in md
        DataDiagnosis().run(test_raw_data, test_rule_file, test_baseline_file, str(self.parent_path), 'md', round=2)
        assert (Path(self.output_md_file).is_file())
        expected_md_file = str(self.parent_path / '../data/diagnosis_summary.md')
        with open(expected_md_file, 'r') as f:
            expect_result = f.read()
        with open(self.output_md_file, 'r') as f:
            summary = f.read()
        assert (summary == expect_result)
        # Test - output in html
        DataDiagnosis().run(test_raw_data, test_rule_file, test_baseline_file, str(self.parent_path), 'html', round=2)
        assert (Path(self.output_html_file).is_file())
        expected_html_file = str(self.parent_path / '../data/diagnosis_summary.html')
        with open(expected_html_file, 'r') as f:
            expect_result = f.read()
        with open(self.output_html_file, 'r') as f:
            summary = f.read()
        assert (summary == expect_result)
        # Test - output all nodes results
        DataDiagnosis().run(
            test_raw_data, test_rule_file, test_baseline_file, str(self.parent_path), 'json', output_all=True
        )
        assert (Path(self.output_all_json_file).is_file())
        expected_result_file = str(self.parent_path / '../data/diagnosis_summary.json')
        with Path(self.output_all_json_file).open() as f:
            data_not_accept_read_from_json = f.read()
        with Path(expected_result_file).open() as f:
            expect_result = f.read()
        assert (data_not_accept_read_from_json == expect_result)

    def test_data_diagnosis_run_without_baseline(self):
        """Test for the run process of rule-based data diagnosis."""
        test_raw_data = str(self.parent_path / 'test_results.jsonl')
        test_rule_file = str(self.parent_path / 'test_rules_without_baseline.yaml')
        test_baseline_file = None

        # Test - output in excel
        DataDiagnosis().run(test_raw_data, test_rule_file, test_baseline_file, str(self.parent_path), 'excel')
        assert (Path(self.output_excel_file).is_file())

        # Test - output in json
        DataDiagnosis().run(test_raw_data, test_rule_file, test_baseline_file, str(self.parent_path), 'json')
        assert (Path(self.output_json_file).is_file())

        # Test - output in jsonl
        DataDiagnosis().run(test_raw_data, test_rule_file, test_baseline_file, str(self.parent_path), 'jsonl')
        assert (Path(self.output_jsonl_file).is_file())

        # Test - output in md
        DataDiagnosis().run(test_raw_data, test_rule_file, test_baseline_file, str(self.parent_path), 'md', round=2)
        assert (Path(self.output_md_file).is_file())

        # Test - output in html
        DataDiagnosis().run(test_raw_data, test_rule_file, test_baseline_file, str(self.parent_path), 'html', round=2)
        assert (Path(self.output_html_file).is_file())

        # Test - output all nodes results
        DataDiagnosis().run(
            test_raw_data, test_rule_file, test_baseline_file, str(self.parent_path), 'json', output_all=True
        )
        assert (Path(self.output_all_json_file).is_file())

    def test_mutli_rules(self):
        """Test multi rules check feature."""
        diag1 = DataDiagnosis()
        # test _check_and_format_rules
        false_rules = [
            {
                'criteria': 'lambda x:x>0',
                'categories': 'KernelLaunch',
                'store': 'true',
                'metrics': ['kernel-launch/event_overhead:\\d+']
            }
        ]
        metric = 'kernel-launch/event_overhead:0'
        for rules in false_rules:
            self.assertRaises(Exception, diag1._check_and_format_rules, rules, metric)
        # Positive case
        true_rules = [
            {
                'categories': 'KernelLaunch',
                'criteria': 'lambda x:x>0.05',
                'store': True,
                'function': 'variance',
                'metrics': ['kernel-launch/event_overhead:\\d+']
            }, {
                'categories': 'CNN',
                'function': 'multi_rules',
                'criteria': 'lambda label:True if label["rule1"]+label["rule2"]>=2 else False'
            }
        ]
        for rules in true_rules:
            assert (diag1._check_and_format_rules(rules, metric))
        # test _run_diagnosis_rules_for_single_node
        rules = {
            'superbench': {
                'rules': {
                    'rule1': {
                        'categories': 'CNN',
                        'criteria': 'lambda x:x<-0.5',
                        'store': True,
                        'function': 'variance',
                        'metrics': ['mem-bw/D2H_Mem_BW']
                    },
                    'rule2': {
                        'categories': 'CNN',
                        'criteria': 'lambda x:x<-0.5',
                        'function': 'variance',
                        'store': True,
                        'metrics': ['kernel-launch/wall_overhead']
                    },
                    'rule3': {
                        'categories': 'CNN',
                        'function': 'multi_rules',
                        'criteria': 'lambda label:True if label["rule1"]+label["rule2"]>=2 else False'
                    }
                }
            }
        }
        baseline = {
            'kernel-launch/wall_overhead': 0.01026,
            'mem-bw/D2H_Mem_BW': 24.3,
        }

        data = {'kernel-launch/wall_overhead': [0.005, 0.005], 'mem-bw/D2H_Mem_BW': [25, 10]}
        diag1._raw_data_df = pd.DataFrame(data, index=['sb-validation-04', 'sb-validation-05'])
        diag1._benchmark_metrics_dict = diag1._get_metrics_by_benchmarks(list(diag1._raw_data_df.columns))
        diag1._parse_rules_and_baseline(rules, baseline)
        (details_row, summary_data_row) = diag1._run_diagnosis_rules_for_single_node('sb-validation-04')
        assert (not details_row)
        (details_row, summary_data_row) = diag1._run_diagnosis_rules_for_single_node('sb-validation-05')
        assert (details_row)
        assert ('CNN' in details_row[0])
        assert (
            details_row[1] == 'kernel-launch/wall_overhead(B/L: 0.0103 VAL: 0.0050 VAR: -51.27% Rule:lambda x:x<-0.5),'
            + 'mem-bw/D2H_Mem_BW(B/L: 24.3000 VAL: 10.0000 VAR: -58.85% Rule:lambda x:x<-0.5),' +
            'rule3:lambda label:True if label["rule1"]+label["rule2"]>=2 else False'
        )

        # Test multi-rule using values of metrics in criteria lambda expression
        diag1 = DataDiagnosis()
        # test _run_diagnosis_rules_for_single_node
        rules = {
            'superbench': {
                'rules': {
                    'rule1': {
                        'categories':
                        'NCCL_DIS',
                        'store':
                        True,
                        'metrics': [
                            'nccl-bw:allreduce-run0/allreduce_1073741824_busbw',
                            'nccl-bw:allreduce-run1/allreduce_1073741824_busbw',
                            'nccl-bw:allreduce-run2/allreduce_1073741824_busbw'
                        ]
                    },
                    'rule2': {
                        'categories': 'NCCL_DIS',
                        'criteria': 'lambda label:True if min(label["rule1"].values())' + '/' +
                        'max(label["rule1"].values())<0.95 else False',
                        'function': 'multi_rules'
                    }
                }
            }
        }

        baseline = {}
        data = {
            'nccl-bw:allreduce-run0/allreduce_1073741824_busbw': [10, 22, 10],
            'nccl-bw:allreduce-run1/allreduce_1073741824_busbw': [23, 23, np.nan],
            'nccl-bw:allreduce-run2/allreduce_1073741824_busbw': [22, 22, np.nan]
        }
        diag1._raw_data_df = pd.DataFrame(data, index=['sb-validation-04', 'sb-validation-05', 'sb-validation-06'])
        diag1._benchmark_metrics_dict = diag1._get_metrics_by_benchmarks(list(diag1._raw_data_df.columns))
        diag1._parse_rules_and_baseline(rules, baseline)
        (details_row, summary_data_row) = diag1._run_diagnosis_rules_for_single_node('sb-validation-04')
        assert (details_row)
        assert ('NCCL_DIS' in details_row[0])
        (details_row, summary_data_row) = diag1._run_diagnosis_rules_for_single_node('sb-validation-05')
        assert (not details_row)
        (details_row, summary_data_row) = diag1._run_diagnosis_rules_for_single_node('sb-validation-06')
        assert (not details_row)

    def test_failure_check(self):
        """Test failure test check feature."""
        diag1 = DataDiagnosis()
        # test _run_diagnosis_rules_for_single_node
        rules = {
            'superbench': {
                'rules': {
                    'rule1': {
                        'categories':
                        'FailedTest',
                        'criteria':
                        'lambda x:x!=0',
                        'function':
                        'failure_check',
                        'metrics': [
                            'gemm-flops/return_code:0', 'gemm-flops/return_code:1', 'gemm-flops/return_code:2',
                            'resnet_models/pytorch-resnet152/return_code'
                        ]
                    }
                }
            }
        }

        baseline = {}

        data = {
            'gemm-flops/return_code:0': [0, -1],
            'gemm-flops/return_code:1': [0, pd.NA],
            'resnet_models/pytorch-resnet152/return_code': [0, -1]
        }
        diag1._raw_data_df = pd.DataFrame(data, index=['sb-validation-04', 'sb-validation-05'])
        diag1._benchmark_metrics_dict = diag1._get_metrics_by_benchmarks(list(diag1._raw_data_df.columns))
        diag1._parse_rules_and_baseline(rules, baseline)
        (details_row, summary_data_row) = diag1._run_diagnosis_rules_for_single_node('sb-validation-04')
        assert (details_row)
        assert ('FailedTest' in details_row[0])
        assert (details_row[1] == 'gemm-flops/return_code:2_miss')
        (details_row, summary_data_row) = diag1._run_diagnosis_rules_for_single_node('sb-validation-05')
        assert (details_row)
        assert ('FailedTest' in details_row[0])
        assert (
            details_row[1] == 'gemm-flops/return_code:0(VAL: -1.0000 Rule:lambda x:x!=0),' +
            'gemm-flops/return_code:1_miss,' + 'gemm-flops/return_code:2_miss,' +
            'resnet_models/pytorch-resnet152/return_code(VAL: -1.0000 Rule:lambda x:x!=0)'
        )