<commit_before># -*- coding:utf-8 -*-
# vim: tabstop=4 shiftwidth=4 softtabstop=4

# Copyright (c) 2012 Samsung SDS Co., LTD
# All Rights Reserved
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

import time
import datetime
import uuid
import unittest

from boto.ec2 import regioninfo
from boto.ec2.cloudwatch import CloudWatchConnection
from boto.ec2.cloudwatch.alarm import MetricAlarm
from boto.exception import BotoServerError
import random 
from synaps import exception
from synaps import utils
from datetime import timedelta

ASYNC_WAIT = 3

class SynapsTestCase(unittest.TestCase):
    def setUp(self):
        access_key = 'changeme'
        secret_key = 'changeme'
        self.synaps = CloudWatchConnection(
            # oss key pair
            aws_access_key_id=access_key, aws_secret_access_key=secret_key,
            is_secure=False, port=8776, path='/monitor',
            region=regioninfo.RegionInfo(None, 'Test Region', 'localhost'),
        )
        
        self.namespace = "SPCS/SYNAPSTEST"
        self.metric_name = "test_metric"
        self.dimensions = {'instance_name':'test instance'}
        
    def tearDown(self):
        pass

    def generate_random_name(self, prefix=""):
        return prefix + str(uuid.uuid4())
        
class ShortCase(SynapsTestCase):
    def test_delete_alarms(self):
        alarmnames = map(self.generate_random_name, ["TEST_ALARM_"] * 10)
        for name in alarmnames:
            alarm = MetricAlarm(name=name, metric=self.metric_name,
                                namespace=self.namespace, statistic="Average",
                                comparison="<", threshold=2.0, period=300,
                                evaluation_periods=2, unit="Percent",
                                description=None, dimensions=self.dimensions,
                                alarm_actions=None,
                                insufficient_data_actions=None,
                                ok_actions=None)
            self.synaps.put_metric_alarm(alarm)

        time.sleep(ASYNC_WAIT * 2)
        
        for alarm in alarmnames:
            self.synaps.delete_alarms(alarms=[alarm])
        
        time.sleep(ASYNC_WAIT)
        
        alarms = self.synaps.describe_alarms()
        for a in alarms:
            self.assertFalse(a.name in alarmnames)

        # should raise 404 exception
        try:
            self.synaps.delete_alarms(alarms=["never_exist_alarmname"])
        except BotoServerError as ex:
            self.assertEqual(ex.error_code, u'404')
        else:
            self.fail("should return 404 error") 
        
    def test_describe_alarm_history(self):
        history_list = self.synaps.describe_alarm_history(
            history_item_type="StateUpdate"
        )
        
        for history in history_list:
            self.assertTrue(history.tem_type, "StateUpdate")
        
    def test_describe_alarms(self):
        for i in range(10):
            alarm = MetricAlarm(name="TEST_ALARM_%02d" % i,
                metric=self.metric_name, namespace=self.namespace,
                statistic="Average", comparison="<", threshold=2.0 * i,
                period=300, evaluation_periods=2, unit="Percent",
                description=None, dimensions=self.dimensions,
                alarm_actions=None, insufficient_data_actions=None,
                ok_actions=None)
            self.synaps.put_metric_alarm(alarm)
                
        alarms = self.synaps.describe_alarms(max_records=3)
        self.assertEqual(len(alarms), 3)

        alarms = self.synaps.describe_alarms(max_records=9,
                                             alarm_name_prefix=u"TEST_")
        self.assertTrue(len(alarms) <= 9)
        for a in alarms:
            self.assertTrue(a.name.startswith("TEST_"))

        alarm_names = ["TEST_ALARM_01", "TEST_ALARM_02"]
        alarms = self.synaps.describe_alarms(max_records=9,
                                             alarm_names=["TEST_ALARM_01"])
        for a in alarms:
            self.assertTrue(a.name in alarm_names)

    
    def test_describe_alarms_for_metric(self):
        for i in range(10):
            alarm = MetricAlarm(name="TEST_ALARM_%02d" % i,
                metric=self.metric_name, namespace=self.namespace,
                statistic="Average", comparison="<", threshold=2.0 * i,
                period=300, evaluation_periods=2, unit="Percent",
                description=None, dimensions=self.dimensions,
                alarm_actions=None, insufficient_data_actions=None,
                ok_actions=None)
            self.synaps.put_metric_alarm(alarm)

        alarms = self.synaps.describe_alarms_for_metric(
            namespace=self.namespace,
            metric_name=self.metric_name,
            dimensions=self.dimensions
        )
    
    def test_get_metric_statistics(self):
        now = datetime.datetime.utcnow()
        now_idx = now.replace(second=0, microsecond=0)
        start_time = now - datetime.timedelta(minutes=30)
        end_time = now

        # input metric
        metric_name = self.generate_random_name("test_metric_")
        dimensions = {self.generate_random_name("key_"):
                      self.generate_random_name("value_")}
        
        values = [10.0, 8.0, 2 ** -32, 0, 50.3, 2 ** 32, -2 ** 32, 4,
                  1000, -3000]
        keys = [now_idx - datetime.timedelta(minutes=i) 
                for i in reversed(range(len(values)))]
        data = zip(keys, values)

        for ts, v in data:
            if v == 0: continue
            self.synaps.put_metric_data(namespace="Test", name=metric_name,
                                        value=v, unit="Percent",
                                        dimensions=dimensions, timestamp=ts)

        time.sleep(ASYNC_WAIT)
        
        period = 4
        stat = self.synaps.get_metric_statistics(
            period=period * 60, start_time=keys[0], end_time=keys[-1],
            metric_name=metric_name, namespace="Test",
            statistics=['Sum', 'Average', 'SampleCount'],
            dimensions=dimensions,
        )
        
        timestamps = keys[::period]

        self.assertEqual([1, 3, 4],
                         map(lambda x:x.get('SampleCount'), stat))
        self.assertEqual(timestamps,
                         map(lambda x:x.get('Timestamp'), stat))
        for e, r in zip([values[0], sum(values[1:5]) / 3,
                         sum(values[5:9]) / 4],<commit_msg>Update testcase<commit_after># -*- coding:utf-8 -*-
# vim: tabstop=4 shiftwidth=4 softtabstop=4

# Copyright (c) 2012 Samsung SDS Co., LTD
# All Rights Reserved
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

import time
import datetime
import uuid
import unittest

from boto.ec2 import regioninfo
from boto.ec2.cloudwatch import CloudWatchConnection
from boto.ec2.cloudwatch.alarm import MetricAlarm
from boto.exception import BotoServerError
import random 
from synaps import exception
from synaps import utils
from datetime import timedelta

ASYNC_WAIT = 3

class SynapsTestCase(unittest.TestCase):
    def setUp(self):
        access_key = 'changeme'
        secret_key = 'changeme'
        self.synaps = CloudWatchConnection(
            # oss key pair
            aws_access_key_id=access_key, aws_secret_access_key=secret_key,
            is_secure=False, port=8776, path='/monitor',
            region=regioninfo.RegionInfo(None, 'Test Region', 'localhost'),
        )
        
        self.namespace = "SPCS/SYNAPSTEST"
        self.metric_name = "test_metric"
        self.dimensions = {'instance_name':'test instance'}
        
    def tearDown(self):
        pass

    def generate_random_name(self, prefix=""):
        return prefix + str(uuid.uuid4())
        
class ShortCase(SynapsTestCase):
    def test_delete_alarms(self):
        alarmnames = map(self.generate_random_name, ["TEST_ALARM_"] * 10)
        for name in alarmnames:
            alarm = MetricAlarm(name=name, metric=self.metric_name,
                                namespace=self.namespace, statistic="Average",
                                comparison="<", threshold=2.0, period=300,
                                evaluation_periods=2, unit="Percent",
                                description=None, dimensions=self.dimensions,
                                alarm_actions=None,
                                insufficient_data_actions=None,
                                ok_actions=None)
            self.synaps.put_metric_alarm(alarm)

        time.sleep(ASYNC_WAIT * 2)
        
        for alarm in alarmnames:
            self.synaps.delete_alarms(alarms=[alarm])
        
        time.sleep(ASYNC_WAIT)
        
        alarms = self.synaps.describe_alarms()
        for a in alarms:
            self.assertFalse(a.name in alarmnames)

        # should raise 404 exception
        try:
            self.synaps.delete_alarms(alarms=["never_exist_alarmname"])
        except BotoServerError as ex:
            self.assertEqual(ex.error_code, u'404')
        else:
            self.fail("should return 404 error") 
        
    def test_describe_alarm_history(self):
        history_list = self.synaps.describe_alarm_history(
            history_item_type="StateUpdate"
        )
        
        for history in history_list:
            self.assertTrue(history.tem_type, "StateUpdate")
        
    def test_describe_alarms(self):
        for i in range(10):
            alarm = MetricAlarm(name="TEST_ALARM_%02d" % i,
                metric=self.metric_name, namespace=self.namespace,
                statistic="Average", comparison="<", threshold=2.0 * i,
                period=300, evaluation_periods=2, unit="Percent",
                description=None, dimensions=self.dimensions,
                alarm_actions=None, insufficient_data_actions=None,
                ok_actions=None)
            self.synaps.put_metric_alarm(alarm)
                
        alarms = self.synaps.describe_alarms(max_records=3)
        self.assertEqual(len(alarms), 3)

        alarms = self.synaps.describe_alarms(max_records=9,
                                             alarm_name_prefix=u"TEST_")
        self.assertTrue(len(alarms) <= 9)
        for a in alarms:
            self.assertTrue(a.name.startswith("TEST_"))

        alarm_names = ["TEST_ALARM_01", "TEST_ALARM_02"]
        alarms = self.synaps.describe_alarms(max_records=9,
                                             alarm_names=["TEST_ALARM_01"])
        for a in alarms:
            self.assertTrue(a.name in alarm_names)

    
    def test_describe_alarms_for_metric(self):
        for i in range(10):
            alarm = MetricAlarm(name="TEST_ALARM_%02d" % i,
                metric=self.metric_name, namespace=self.namespace,
                statistic="Average", comparison="<", threshold=2.0 * i,
                period=300, evaluation_periods=2, unit="Percent",
                description=None, dimensions=self.dimensions,
                alarm_actions=None, insufficient_data_actions=None,
                ok_actions=None)
            self.synaps.put_metric_alarm(alarm)

        alarms = self.synaps.describe_alarms_for_metric(
            namespace=self.namespace,
            metric_name=self.metric_name,
            dimensions=self.dimensions
        )
    
    def test_get_metric_statistics(self):
        now = datetime.datetime.utcnow()
        now_idx = now.replace(second=0, microsecond=0)
        start_time = now - datetime.timedelta(minutes=30)
        end_time = now

        # input metric
        metric_name = self.generate_random_name("test_metric_")
        dimensions = {self.generate_random_name("key_"):
                      self.generate_random_name("value_")}
        
        values = [10.0, 8.0, 2 ** -32, 0, 50.3, 2 ** 16, -2 ** 16, 4,
                  1000, -3000]
        keys = [now_idx - datetime.timedelta(minutes=i) 
                for i in reversed(range(len(values)))]
        data = zip(keys, values)

        for ts, v in data:
            if v == 0: continue
            self.synaps.put_metric_data(namespace="Test", name=metric_name,
                                        value=v, unit="Percent",
                                        dimensions=dimensions, timestamp=ts)

        time.sleep(ASYNC_WAIT)
        
        period = 4
        stat = self.synaps.get_metric_statistics(
            period=period * 60, start_time=keys[0], end_time=keys[-1],
            metric_name=metric_name, namespace="Test",
            statistics=['Sum', 'Average', 'SampleCount'],
            dimensions=dimensions,
        )
        
        timestamps = keys[::period]

        self.assertEqual([1, 3, 4],
                         map(lambda x:x.get('SampleCount'), stat))
        self.assertEqual(timestamps,
                         map(lambda x:x.get('Timestamp'), stat))
        for e, r in zip([values[0], sum(values[1:5]) / 3,
                         sum(values[5:9]) / 4],
{'fraction_non_alphanumeric': 0.09551684751570531, 'fraction_numerical': 0.01670474014848658, 'mean_word_length': 2.033564313555652, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '516574', 'n_tokens_mistral': 4141, 'n_tokens_neox': 3932, 'n_words': 838}