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

"""
lg stats basic offline (not requiring the ROS node itself to run) tests.

Run just this test module manually:
    cd lg_ros_nodes  # must be run from here due to nosetest cfg file
    nosetests --verbosity=3 -s -l DEBUG catkin/src/lg_stats/test/offline/test_lg_stats_basic.py

    nosetestst is troublesome and ignores pytest stuff (e.g. decorators),
        investigate with rosunit <test_module.py>

"""


import os
import time
import json

import pytest
import rospkg

from lg_common.helpers import write_log_to_file
from std_msgs.msg import Bool
from interactivespaces_msgs.msg import GenericMessage
from appctl.msg import Mode
from lg_stats.msg import Session
from lg_stats.msg import Event
from lg_stats import ROS_NODE_NAME
from lg_stats import Processor
from lg_stats import Submitter
from lg_stats import InfluxDirect
from lg_stats import InfluxTelegraf
from lg_stats import InfluxMock
from lg_stats import EmptyIncomingMessage


real_in_msg_director_scene = """
{
"type": "json",
"message": {
  "description": "scene-3-1 desc",
  "duration": 2,
  "name": "scene-3-1",
  "resource_uri": "/director_api/scene/bbb94866-2216-41a2-83b4-13ba35a3e9dc__scene-3-1/",
  "slug": "bbb94866-2216-41a2-83b4-13ba35a3e9dc__scene-3-1",
  "windows": []
  }
}
"""


class MockTopicPublisher(object):
    def __init__(self):
        self.messages = []

    def publish(self, msg):
        self.messages.append(msg)


class TestSubmitters(object):

    def test_submitters_instantiation(self):
        pytest.raises(RuntimeError, Submitter)
        InfluxDirect()
        InfluxTelegraf()
        InfluxMock()

    def test_get_data_for_influx(self):
        # prepare message for testing, as if it was received (incoming message)
        slot = json.loads(real_in_msg_director_scene)["message"]
        in_msg = GenericMessage(type="json", message=json.dumps(slot))
        # get outgoing message (to /lg_stats/debug) - via Processor instance
        p = Processor(watched_topic="/director/scene",
                      msg_slot="message",
                      watched_field_name="slug")
        out_msg = p.get_outbound_message(in_msg)
        whole_field_name = "message.slug"  # depends on the particular message type above
        # now finally test the method of submitters:
        influx = InfluxDirect.get_data_for_influx(out_msg)
        assert isinstance(influx, dict)
        assert influx["measurement"] == p.watched_topic
        assert influx["tags"]["field_name"] == whole_field_name
        assert influx["tags"]["value"] == "bbb94866-2216-41a2-83b4-13ba35a3e9dc__scene-3-1"
        influx = InfluxTelegraf.get_data_for_influx(out_msg)
        assert isinstance(influx, str)
        assert influx.startswith(p.watched_topic)
        assert influx.find("field_name=%s" % whole_field_name) > -1
        assert influx.find("value=%s" % out_msg.value) > -1
        # this mock just calls InfluxTelegraf - do the same assertions
        influx = InfluxMock.get_data_for_influx(out_msg)
        assert isinstance(influx, str)
        assert influx.startswith(p.watched_topic)
        assert influx.find("field_name=%s" % whole_field_name) > -1
        assert influx.find("value=%s" % out_msg.value) > -1


class TestLGStatsProcessor(object):
    """
    Test Processor class.
    It's basically offline test, no ROS interaction.

    """
    def test_basic(self):
        p = Processor(resolution=200)
        assert p.resolution == 200

    def test_whole_field_name(self):
        p = Processor(watched_topic="/director/scene",
                      msg_slot="message",
                      watched_field_name="slug")
        assert p.get_whole_field_name() == "message.slug"
        p = Processor(watched_topic="/director/scene",
                      watched_field_name="somethingelse")
        assert p.get_whole_field_name() == "somethingelse"

    def test_publish(self):
        pub = MockTopicPublisher()
        p = Processor(watched_topic=None,
                      watched_field_name="application",
                      debug_pub=pub,
                      resolution=200)
        msg1 = Session(application="someapplication1")
        out = p.get_outbound_message(msg1)
        p.debug_pub.publish(out)
        assert isinstance(pub.messages[0], Event)<commit_msg>fix quotes in the test, touch: #126<commit_after>#!/usr/bin/env python

"""
lg stats basic offline (not requiring the ROS node itself to run) tests.

Run just this test module manually:
    cd lg_ros_nodes  # must be run from here due to nosetest cfg file
    nosetests --verbosity=3 -s -l DEBUG catkin/src/lg_stats/test/offline/test_lg_stats_basic.py

    nosetestst is troublesome and ignores pytest stuff (e.g. decorators),
        investigate with rosunit <test_module.py>

"""


import os
import time
import json

import pytest
import rospkg

from lg_common.helpers import write_log_to_file
from std_msgs.msg import Bool
from interactivespaces_msgs.msg import GenericMessage
from appctl.msg import Mode
from lg_stats.msg import Session
from lg_stats.msg import Event
from lg_stats import ROS_NODE_NAME
from lg_stats import Processor
from lg_stats import Submitter
from lg_stats import InfluxDirect
from lg_stats import InfluxTelegraf
from lg_stats import InfluxMock
from lg_stats import EmptyIncomingMessage


real_in_msg_director_scene = """
{
"type": "json",
"message": {
  "description": "scene-3-1 desc",
  "duration": 2,
  "name": "scene-3-1",
  "resource_uri": "/director_api/scene/bbb94866-2216-41a2-83b4-13ba35a3e9dc__scene-3-1/",
  "slug": "bbb94866-2216-41a2-83b4-13ba35a3e9dc__scene-3-1",
  "windows": []
  }
}
"""


class MockTopicPublisher(object):
    def __init__(self):
        self.messages = []

    def publish(self, msg):
        self.messages.append(msg)


class TestSubmitters(object):

    def test_submitters_instantiation(self):
        pytest.raises(RuntimeError, Submitter)
        InfluxDirect()
        InfluxTelegraf()
        InfluxMock()

    def test_get_data_for_influx(self):
        # prepare message for testing, as if it was received (incoming message)
        slot = json.loads(real_in_msg_director_scene)["message"]
        in_msg = GenericMessage(type="json", message=json.dumps(slot))
        # get outgoing message (to /lg_stats/debug) - via Processor instance
        p = Processor(watched_topic="/director/scene",
                      msg_slot="message",
                      watched_field_name="slug")
        out_msg = p.get_outbound_message(in_msg)
        whole_field_name = "message.slug"  # depends on the particular message type above
        # now finally test the method of submitters:
        influx = InfluxDirect.get_data_for_influx(out_msg)
        assert isinstance(influx, dict)
        assert influx["measurement"] == p.watched_topic
        assert influx["tags"]["field_name"] == whole_field_name
        assert influx["tags"]["value"] == "bbb94866-2216-41a2-83b4-13ba35a3e9dc__scene-3-1"
        influx = InfluxTelegraf.get_data_for_influx(out_msg)
        assert isinstance(influx, str)
        assert influx.startswith(p.watched_topic)
        assert influx.find("field_name=\"%s\"" % whole_field_name) > -1
        assert influx.find("value=\"%s\"" % out_msg.value) > -1
        # this mock just calls InfluxTelegraf - do the same assertions
        influx = InfluxMock.get_data_for_influx(out_msg)
        assert isinstance(influx, str)
        assert influx.startswith(p.watched_topic)
        assert influx.find("field_name=\"%s\"" % whole_field_name) > -1
        assert influx.find("value=\"%s\"" % out_msg.value) > -1


class TestLGStatsProcessor(object):
    """
    Test Processor class.
    It's basically offline test, no ROS interaction.

    """
    def test_basic(self):
        p = Processor(resolution=200)
        assert p.resolution == 200

    def test_whole_field_name(self):
        p = Processor(watched_topic="/director/scene",
                      msg_slot="message",
                      watched_field_name="slug")
        assert p.get_whole_field_name() == "message.slug"
        p = Processor(watched_topic="/director/scene",
                      watched_field_name="somethingelse")
        assert p.get_whole_field_name() == "somethingelse"

    def test_publish(self):
        pub = MockTopicPublisher()
        p = Processor(watched_topic=None,
                      watched_field_name="application",
                      debug_pub=pub,
                      resolution=200)
        msg1 = Session(application="someapplication1")
        out = p.get_outbound_message(msg1)
        p.debug_pub.publish(out)
        assert isinstance(pub.messages[0], Event)
{'fraction_non_alphanumeric': 0.10006909258406264, 'fraction_numerical': 0.021533855366190697, 'mean_word_length': 3.071729957805907, 'pattern_counts': {'":': 16, '<': 5, '<?xml version=': 0, '>': 13, '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': '6950493', 'n_tokens_mistral': 2902, 'n_tokens_neox': 2730, 'n_words': 605}