hastic-server/analytics/tests/test_detectors.py

import unittest
import pandas as pd

from detectors import pattern_detector, threshold_detector, anomaly_detector

class TestPatternDetector(unittest.TestCase):

    def test_small_dataframe(self):

        data = [[0,1], [1,2]]
        dataframe = pd.DataFrame(data, columns=['timestamp', 'values'])
        cache = {'windowSize': 10}

        detector = pattern_detector.PatternDetector('GENERAL', 'test_id')
        
        with self.assertRaises(ValueError):
            detector.detect(dataframe, cache)


class TestThresholdDetector(unittest.TestCase):

    def test_invalid_cache(self):

        detector = threshold_detector.ThresholdDetector()
        
        with self.assertRaises(ValueError):
            detector.detect([], None)

        with self.assertRaises(ValueError):
            detector.detect([], {})


class TestAnomalyDetector(unittest.TestCase):

    def test_dataframe(self):
        data_val = [0, 1, 2, 1, 2, 10, 1, 2, 1]
        data_ind = [1523889000000 + i for i in range(len(data_val))]
        data = {'timestamp': data_ind, 'value': data_val}
        dataframe = pd.DataFrame(data = data)
        dataframe['timestamp'] = pd.to_datetime(dataframe['timestamp'], unit='ms')
        cache =  {
            'confidence': 2,
            'alpha': 0.1,
        }
        detector = anomaly_detector.AnomalyDetector()
        detect_result = detector.detect(dataframe, cache)
        result = [(1523889000005.0, 1523889000005.0)]
        self.assertEqual(result, detect_result['segments'])