<commit_before>#!/usr/bin/env python

#
# This file is used to test host- and service-downtimes.
#

import sys
import time
import datetime
import os
import string
import re
import random
import unittest
import copy

# import the shinken library from the parent directory
import __import_shinken ; del __import_shinken

import shinken
from shinken.objects.config import Config
from shinken.objects.command import Command
from shinken.objects.module import Module

from shinken.dispatcher import Dispatcher
from shinken.log import logger
from shinken.modulesctx import modulesctx
from shinken.scheduler import Scheduler
from shinken.macroresolver import MacroResolver
from shinken.external_command import ExternalCommandManager, ExternalCommand
from shinken.check import Check
from shinken.message import Message
from shinken.arbiterlink import ArbiterLink
from shinken.schedulerlink import SchedulerLink
from shinken.pollerlink import PollerLink
from shinken.reactionnerlink import ReactionnerLink
from shinken.brokerlink import BrokerLink
from shinken.satellitelink import SatelliteLink
from shinken.notification import Notification
from shinken.modulesmanager import ModulesManager
from shinken.basemodule import BaseModule

from shinken.brok import Brok

from shinken.daemons.schedulerdaemon import Shinken
from shinken.daemons.brokerdaemon import Broker
from shinken.daemons.arbiterdaemon import Arbiter

# Modules are by default on the ../modules
myself = os.path.abspath(__file__)

global modules_dir
modules_dir = "modules"

def define_modules_dir(val):
    global modules_dir
    modules_dir = val

class __DUMMY:
    def add(self, obj):
        pass

logger.load_obj(__DUMMY())



# We overwrite the functions time() and sleep()
# This way we can modify sleep() so that it immediately returns although
# for a following time() it looks like thee was actually a delay.
# This massively speeds up the tests.
class TimeHacker(object):

    def __init__(self):
        self.my_offset = 0
        self.my_starttime = time.time()
        self.my_oldtime = time.time
        self.original_time_time = time.time
        self.original_time_sleep = time.sleep
        self.in_real_time = True

    def my_time_time(self):
        return self.my_oldtime() + self.my_offset

    def my_time_sleep(self, delay):
        self.my_offset += delay

    def time_warp(self, duration):
        self.my_offset += duration

    def set_my_time(self):
        if self.in_real_time:
            time.time = self.my_time_time
            time.sleep = self.my_time_sleep
            self.in_real_time = False

# If external processes or time stamps for files are involved, we must
# revert the fake timing routines, because these externals cannot be fooled.
# They get their times from the operating system.
    def set_real_time(self):
        if not self.in_real_time:
            time.time = self.original_time_time
            time.sleep = self.original_time_sleep
            self.in_real_time = True


#Time hacking for every test!
time_hacker = TimeHacker()
time_hacker.set_my_time()


class Pluginconf(object):
    pass


class _Unittest2CompatMixIn:
    """
    Mixin for simulating methods new in unittest2 resp. Python 2.7.

    Every test-case should inherit this *after* unittest.TestCase to
    make the compatiblity-methods available if they are not defined in
    unittest.TestCase already. Example::

       class MyTestCase(unittest.TestCase, Unittest2CompatMixIn):
           ...
    In our case, it's better to always inherit from ShinkenTest

    """
    def assertNotIn(self, member, container, msg=None):
       self.assertTrue(member not in container)

    def assertIn(self, member, container, msg=None):
        self.assertTrue(member in container)

    def assertIsInstance(self, obj, cls, msg=None):
        self.assertTrue(isinstance(obj, cls))

    def assertRegexpMatches(self, line, pattern):
        r = re.search(pattern, line)
        self.assertTrue(r is not None)

    def assertIs(self, obj, cmp, msg=None):
        self.assertTrue(obj is cmp)


class ShinkenTest(unittest.TestCase, Unittest2CompatMixIn):
    def setUp(self):
        self.setup_with_file('etc/shinken_1r_1h_1s.cfg')

    def setup_with_file(self, path):
        # i am arbiter-like
        self.broks = {}
        self.me = None
        self.log = logger
        self.log.load_obj(self)
        self.config_files = [path]
        self.conf = Config()
        buf = self.conf.read_config(self.config_files)
        raw_objects = self.conf.read_config_buf(buf)
        self.conf.create_objects_for_type(raw_objects, 'arbiter')
        self.conf.create_objects_for_type(raw_objects, 'module')
        self.conf.early_arbiter_linking()
        self.conf.create_objects(raw_objects)
        self.conf.old_properties_names_to_new()<commit_msg>Enh : Test - Keep on merging<commit_after>#!/usr/bin/env python

#
# This file is used to test host- and service-downtimes.
#

import sys
import time
import datetime
import os
import string
import re
import random
import unittest
import copy

# import the shinken library from the parent directory
import __import_shinken ; del __import_shinken

import shinken
from shinken.objects.config import Config
from shinken.objects.command import Command
from shinken.objects.module import Module

from shinken.dispatcher import Dispatcher
from shinken.log import logger
from shinken.modulesctx import modulesctx
from shinken.scheduler import Scheduler
from shinken.macroresolver import MacroResolver
from shinken.external_command import ExternalCommandManager, ExternalCommand
from shinken.check import Check
from shinken.message import Message
from shinken.arbiterlink import ArbiterLink
from shinken.schedulerlink import SchedulerLink
from shinken.pollerlink import PollerLink
from shinken.reactionnerlink import ReactionnerLink
from shinken.brokerlink import BrokerLink
from shinken.satellitelink import SatelliteLink
from shinken.notification import Notification
from shinken.modulesmanager import ModulesManager
from shinken.basemodule import BaseModule

from shinken.brok import Brok

from shinken.daemons.schedulerdaemon import Shinken
from shinken.daemons.brokerdaemon import Broker
from shinken.daemons.arbiterdaemon import Arbiter

# Modules are by default on the ../modules
myself = os.path.abspath(__file__)

global modules_dir
modules_dir = "modules"

def define_modules_dir(val):
    global modules_dir
    modules_dir = val

class __DUMMY:
    def add(self, obj):
        pass

logger.load_obj(__DUMMY())



# We overwrite the functions time() and sleep()
# This way we can modify sleep() so that it immediately returns although
# for a following time() it looks like thee was actually a delay.
# This massively speeds up the tests.
class TimeHacker(object):

    def __init__(self):
        self.my_offset = 0
        self.my_starttime = time.time()
        self.my_oldtime = time.time
        self.original_time_time = time.time
        self.original_time_sleep = time.sleep
        self.in_real_time = True

    def my_time_time(self):
        return self.my_oldtime() + self.my_offset

    def my_time_sleep(self, delay):
        self.my_offset += delay

    def time_warp(self, duration):
        self.my_offset += duration

    def set_my_time(self):
        if self.in_real_time:
            time.time = self.my_time_time
            time.sleep = self.my_time_sleep
            self.in_real_time = False

# If external processes or time stamps for files are involved, we must
# revert the fake timing routines, because these externals cannot be fooled.
# They get their times from the operating system.
    def set_real_time(self):
        if not self.in_real_time:
            time.time = self.original_time_time
            time.sleep = self.original_time_sleep
            self.in_real_time = True


#Time hacking for every test!
time_hacker = TimeHacker()
time_hacker.set_my_time()


class Pluginconf(object):
    pass


class _Unittest2CompatMixIn:
    """
    Mixin for simulating methods new in unittest2 resp. Python 2.7.

    Every test-case should inherit this *after* unittest.TestCase to
    make the compatiblity-methods available if they are not defined in
    unittest.TestCase already. Example::

       class MyTestCase(unittest.TestCase, Unittest2CompatMixIn):
           ...
    In our case, it's better to always inherit from ShinkenTest

    """
    def assertNotIn(self, member, container, msg=None):
       self.assertTrue(member not in container)

    def assertIn(self, member, container, msg=None):
        self.assertTrue(member in container)

    def assertIsInstance(self, obj, cls, msg=None):
        self.assertTrue(isinstance(obj, cls))

    def assertRegexpMatches(self, line, pattern):
        r = re.search(pattern, line)
        self.assertTrue(r is not None)

    def assertIs(self, obj, cmp, msg=None):
        self.assertTrue(obj is cmp)


class ShinkenTest(unittest.TestCase, _Unittest2CompatMixIn):
    def setUp(self):
        self.setup_with_file('etc/shinken_1r_1h_1s.cfg')

    def setup_with_file(self, path):
        # i am arbiter-like
        self.broks = {}
        self.me = None
        self.log = logger
        self.log.load_obj(self)
        self.config_files = [path]
        self.conf = Config()
        buf = self.conf.read_config(self.config_files)
        raw_objects = self.conf.read_config_buf(buf)
        self.conf.create_objects_for_type(raw_objects, 'arbiter')
        self.conf.create_objects_for_type(raw_objects, 'module')
        self.conf.early_arbiter_linking()
        self.conf.create_objects(raw_objects)
        self.conf.old_properties_names_to_new()
{'fraction_non_alphanumeric': 0.0705419942076955, 'fraction_numerical': 0.002068680182043856, 'mean_word_length': 3.5140056022408963, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7421105', 'n_tokens_mistral': 2970, 'n_tokens_neox': 2892, 'n_words': 843}