<commit_before># coding=utf-8
import fnmatch
import os
import platform
import re
import subprocess
import sys
import time
from typing import Any, List, Tuple, cast

from ..util import MonitorConfigurationError
from .monitor import Monitor, register

try:
    import pydbus
except ImportError:
    pydbus = None


@register
class MonitorSvc(Monitor):
    """Monitor a service handled by daemontools."""

    type = "svc"
    path = ""

    def __init__(self, name: str, config_options: dict) -> None:
        super().__init__(name, config_options)
        self.path = cast(str, self.get_config_option("path", required=True))
        self.params = ("svok %s" % self.path).split(" ")

    def run_test(self) -> bool:
        if self.path == "":
            return self.record_fail("Path is not configured")
        try:
            result = subprocess.call(self.params)
            if result is None:
                result = 0
            if result > 0:
                return self.record_fail("svok returned %d" % int(result))
            return self.record_success()
        except Exception as e:
            return self.record_fail("Exception while executing svok: %s" % e)

    def describe(self) -> str:
        return (
            "Checking that the supervise-managed service in %s is running." % self.path
        )

    def get_params(self) -> Tuple:
        return (self.path,)


@register
class MonitorService(Monitor):
    """Monitor a Windows service"""

    service_name = ""
    want_state = "RUNNING"
    host = "."
    type = "service"

    def __init__(self, name: str, config_options: dict) -> None:
        super().__init__(name, config_options)
        self.service_name = self.get_config_option("service", required=True)
        self.want_state = self.get_config_option("state", default="RUNNING")
        self.host = self.get_config_option("host", default=".")

        if self.want_state not in ["RUNNING", "STOPPED"]:
            raise MonitorConfigurationError(
                "invalid state {0} for MonitorService".format(self.want_state)
            )

    def run_test(self) -> bool:
        """Check the service is in the desired state"""
        SVC_STATUS_LINE = 2
        try:
            if platform.system() == "CYGWIN_NT-6.0":
                host = "\\\\\\\\" + self.host
            elif platform.system() in ["Microsoft", "Windows"]:
                host = "\\\\" + self.host
            else:
                # we need windows for sc
                return self.record_fail(
                    "Cannot check for Windows services while running on a non-Windows platform."
                )

            output = str(
                subprocess.check_output(["sc", host, "query", self.service_name, "state=all"])
                , "utf-8")            
            lines = output.split(os.linesep)
            lines = [x for x in lines if len(x) > 0]
            if self.want_state in lines[SVC_STATUS_LINE]:
                return self.record_success()           
        except Exception as e:
            sys.stderr.write("%s\n" % e)
            pass
        return self.record_fail()

    def describe(self) -> str:
        """Explains what this instance is checking"""
        return "checking for service called %s in state %s" % (
            self.service_name,
            self.want_state,
        )

    def get_params(self) -> Tuple:
        return (self.host, self.service_name, self.want_state)


@register
class MonitorRC(Monitor):
    """Monitor a service handled by an rc.d script.

    This monitor checks the return code of /usr/local/etc/rc.d/<name>
    and reports failure if it's non-zero by default.<commit_msg>FIX "Codacy" quality test<commit_after># coding=utf-8
import fnmatch
import os
import platform
import re
import subprocess
import sys
import time
from typing import Any, List, Tuple, cast

from ..util import MonitorConfigurationError
from .monitor import Monitor, register

try:
    import pydbus
except ImportError:
    pydbus = None


@register
class MonitorSvc(Monitor):
    """Monitor a service handled by daemontools."""

    type = "svc"
    path = ""

    def __init__(self, name: str, config_options: dict) -> None:
        super().__init__(name, config_options)
        self.path = cast(str, self.get_config_option("path", required=True))
        self.params = ("svok %s" % self.path).split(" ")

    def run_test(self) -> bool:
        if self.path == "":
            return self.record_fail("Path is not configured")
        try:
            result = subprocess.call(self.params)
            if result is None:
                result = 0
            if result > 0:
                return self.record_fail("svok returned %d" % int(result))
            return self.record_success()
        except Exception as e:
            return self.record_fail("Exception while executing svok: %s" % e)

    def describe(self) -> str:
        return (
            "Checking that the supervise-managed service in %s is running." % self.path
        )

    def get_params(self) -> Tuple:
        return (self.path,)


@register
class MonitorService(Monitor):
    """Monitor a Windows service"""

    service_name = ""
    want_state = "RUNNING"
    host = "."
    type = "service"

    def __init__(self, name: str, config_options: dict) -> None:
        super().__init__(name, config_options)
        self.service_name = self.get_config_option("service", required=True)
        self.want_state = self.get_config_option("state", default="RUNNING")
        self.host = self.get_config_option("host", default=".")

        if self.want_state not in ["RUNNING", "STOPPED"]:
            raise MonitorConfigurationError(
                "invalid state {0} for MonitorService".format(self.want_state)
            )

    def run_test(self) -> bool:
        """Check the service is in the desired state"""
        SVC_STATUS_LINE = 2
        try:
            if platform.system() == "CYGWIN_NT-6.0":
                host = "\\\\\\\\" + self.host
            elif platform.system() in ["Microsoft", "Windows"]:
                host = "\\\\" + self.host
            else:
                # we need windows for sc
                return self.record_fail(
                    "Cannot check for Windows services while running on a non-Windows platform."
                )

            output = str(subprocess.check_output(["sc", host, "query", self.service_name, "state=all"]),"utf-8")
            lines = output.split(os.linesep)
            lines = [x for x in lines if len(x) > 0]
            if self.want_state in lines[SVC_STATUS_LINE]:
                return self.record_success()           
        except Exception as e:
            sys.stderr.write("%s\n" % e)
            pass
        return self.record_fail()

    def describe(self) -> str:
        """Explains what this instance is checking"""
        return "checking for service called %s in state %s" % (
            self.service_name,
            self.want_state,
        )

    def get_params(self) -> Tuple:
        return (self.host, self.service_name, self.want_state)


@register
class MonitorRC(Monitor):
    """Monitor a service handled by an rc.d script.

    This monitor checks the return code of /usr/local/etc/rc.d/<name>
    and reports failure if it's non-zero by default.
{'fraction_non_alphanumeric': 0.11170431211498973, 'fraction_numerical': 0.0024640657084188913, 'mean_word_length': 2.1129100979974433, 'pattern_counts': {'":': 4, '<': 5, '<?xml version=': 0, '>': 25, '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': '502422', 'n_tokens_mistral': 2106, 'n_tokens_neox': 2029, 'n_words': 610}