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

"""
lg stats testing scenarios.

starting roslaunch for development:
    roslaunch --screen lg_stats/test/online/test_lg_stats.test
    could do:
    py.test -s -v lg_stats/test/online/test_lg_stats.py

running tests manually:
    rostest lg_stats/test/online/test_lg_stats.test
        (as long as it contains <test> tag, it's the same as launch file)

"""


import os
import unittest
import time
from multiprocessing import Array

import pytest
import rostest
import rospy
import rospkg
import rostopic

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 LG_STATS_DEBUG_TOPIC_DEFAULT


RESULT = Array('c', 100)  # size


class TestLGStatsRealMessageChain(object):
    """
    Test each of the watched topics.
    Configured via roslanch.xml file.

    """

    def setup_method(self, method):
        RESULT.value = "UNDEFINED"

    @staticmethod
    def callback(msg):
        rospy.loginfo("callback received type: '%s'" % type(msg))
        rospy.loginfo(msg)
        RESULT.value = msg.value

    def test_send_director_scene(self):
        # TODO
        # some stuff will be moved into setup_method / setup_class methods
        # send a scene message which shall result in a reaction on lg_stats/debug topic
        msg = GenericMessage(type="json", message="something")
        pub = rospy.Publisher("/director/scene", GenericMessage, queue_size=3)
        # WARNING: issues retvieving get_param value from the test file, still
        #   getting the default value (regardless of calling it before or after init_node)
        debug_topic = "%s/%s" % (ROS_NODE_NAME,
                                 rospy.get_param("~debug_topic", LG_STATS_DEBUG_TOPIC_DEFAULT))
        rospy.Subscriber(debug_topic, Event, self.callback)
        rospy.init_node(ROS_NODE_NAME, anonymous=True)
        # after this call the ros infrastructure starts up, sending a message right
        # after this results in a lost message sometimes ... wait
        rospy.sleep(2)
        pub.publish(msg)
        # need to wait since message will only be sent out if the state
        # has not changed for time greater than resolution
        rospy.sleep(1.1)
        pub.publish(msg)
        # wait a bit
        for count in range(3):
            if RESULT.value != "UNDEFINED":
                break
            rospy.sleep(1)
        assert RESULT.value == "something"

    def test_send_appctl_mode(self):
        # TODO
        # some stuff will be moved into setup_method / setup_class methods
        # send a scene message which shall result in a reaction on lg_stats/debug topic
        msg = Mode(mode="a good mode")
        pub = rospy.Publisher("/appctl/mode", Mode, queue_size=3)
        # WARNING: issues retvieving get_param value from the test file, still
        #   getting the default value (regardless of calling it before or after init_node)
        debug_topic = "%s/%s" % (ROS_NODE_NAME,
                                 rospy.get_param("~debug_topic", LG_STATS_DEBUG_TOPIC_DEFAULT))
        rospy.Subscriber(debug_topic, Event, self.callback)
        rospy.init_node(ROS_NODE_NAME, anonymous=True)
        # after this call the ros infrastructure starts up, sending a message right
        # after this results in a lost message sometimes ... wait
        rospy.sleep(2)
        pub.publish(msg)
        # need to wait since message will only be sent out if the state
        # has not changed for time greater than resolution
        rospy.sleep(1.1)
        pub.publish(msg)
        # wait a bit
        for count in range(3):
            if RESULT.value != "UNDEFINED":
                break
            rospy.sleep(1)
        assert RESULT.value == "a good mode"

    def test_send_statistics_session(self):
        # TODO
        # some stuff will be moved into setup_method / setup_class methods
        # send a scene message which shall result in a reaction on lg_stats/debug topic
        msg = Session(application="a good application")
        pub = rospy.Publisher("/statistics/session", Session, queue_size=3)
        # WARNING: issues retvieving get_param value from the test file, still
        #   getting the default value (regardless of calling it before or after init_node)
        debug_topic = "%s/%s" % (ROS_NODE_NAME,
                                 rospy.get_param("~debug_topic", LG_STATS_DEBUG_TOPIC_DEFAULT))
        rospy.Subscriber(debug_topic, Event, self.callback)
        rospy.init_node(ROS_NODE_NAME, anonymous=True)
        # after this call the ros infrastructure starts up, sending a message right
        # after this results in a lost message sometimes ... wait
        rospy.sleep(2)
        pub.publish(msg)
        # need to wait since message will only be sent out if the state
        # has not changed for time greater than resolution
        rospy.sleep(1.1)
        pub.publish(msg)
        # wait a bit
        for count in range(3):
            if RESULT.value != "UNDEFINED":
                break
            rospy.sleep(1)
        assert RESULT.value == "a good application"

    def test_send_activity_active(self):
        # TODO
        # some stuff will be moved into setup_method / setup_class methods
        # send a scene message which shall result in a reaction on lg_stats/debug topic
        msg = Bool(data=True)
        pub = rospy.Publisher("/activity/active", Bool, queue_size=3)
        # WARNING: issues retrieving get_param value from the test file, still
        #   getting the default value (regardless of calling it before or after init_node)
        debug_topic = "%s/%s" % (ROS_NODE_NAME,
                                 rospy.get_param("~debug_topic", LG_STATS_DEBUG_TOPIC_DEFAULT))
        rospy.Subscriber(debug_topic, Event, self.callback)
        rospy.init_node(ROS_NODE_NAME, anonymous=True)
        # after this call the ros infrastructure starts up, sending a message right
        # after this results in a lost message sometimes ... wait
        rospy.sleep(2)
        pub.publish(msg)
        # need to wait since message will only be sent out if the state
        # has not changed for time greater than resolution
        rospy.sleep(1.1)
        pub.publish(msg)
        # wait a bit
        for count in range(3):
            if RESULT.value != "UNDEFINED":
                break
            rospy.sleep(1)
        assert RESULT.value == "True"


if __name__ == "__main__":
    # pytest must provide result XML file just as rostest.rosrun would do
    # otherwise: FAILURE: test [test_lg_media_basic] did not generate test results
    test_pkg = ROS_NODE_NAME
    test_name = "test_lg_stats"
    test_dir = os.path.join(rospkg.get_test_results_dir(env=None), test_pkg)
    pytest_result_path = os.path.join(test_dir, "rosunit-%s.xml" % test_name)
    # run only itself
    test_path = os.path.abspath(os.path.abspath(__file__))
    # output is unfortunately handled / controlled by above layer of rostest (-s has no effect)
    pytest.main("%s -s -v --junit-xml=%s" % (test_path, pytest_result_path))<commit_msg>current tests refactored, code reused, shortened, touch: #126<commit_after>#!/usr/bin/env python

"""
lg stats testing scenarios.

starting roslaunch for development:
    roslaunch --screen lg_stats/test/online/test_lg_stats.test
    could do:
    py.test -s -v lg_stats/test/online/test_lg_stats.py

running tests manually:
    rostest lg_stats/test/online/test_lg_stats.test
        (as long as it contains <test> tag, it's the same as launch file)

"""


import os
import unittest
import time
from multiprocessing import Array

import pytest
import rostest
import rospy
import rospkg
import rostopic

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 LG_STATS_DEBUG_TOPIC_DEFAULT


RESULT = Array('c', 100)  # size


class TestLGStatsRealMessageChain(object):
    """
    Test each of the watched topics.
    Configured via roslanch.xml file.

    """

    def setup_method(self, method):
        RESULT.value = "UNDEFINED"
        # subscribed to the /lg_stats/debug topic
        # WARNING: issues retrieving rospy.get_param value from the test file, still
        #   getting the default value (regardless of calling it before or after init_node)
        debug_topic = "%s/%s" % (ROS_NODE_NAME, LG_STATS_DEBUG_TOPIC_DEFAULT)
        rospy.Subscriber(debug_topic, Event, self.callback)

    @staticmethod
    def callback(msg):
        """
        Handler of incoming messages on /lg_stats/debug topic.
        Set the shared mem value based on the received message.

        """
        rospy.loginfo("callback received type: '%s'" % type(msg))
        rospy.loginfo(msg)
        RESULT.value = msg.value

    def checker(self, publisher, msg_to_send, expected_value):
        """
        This method is called after init_node
        Wait a bit until the ros infrastructure starts up, sending a message right
        init_node results in a lost message sometimes ...

        Second sent messages triggers processing of the first one.
        Second messages must be send after 'resolution' time period.

        """
        rospy.sleep(2)
        publisher.publish(msg_to_send)
        # need to wait since message will only be sent out if the state
        # has not changed for time greater than resolution
        rospy.sleep(1.1)
        publisher.publish(msg_to_send)
        # wait a bit, call back shall set the share mem value accordingly
        for count in range(3):
            if RESULT.value != "UNDEFINED":
                break
            rospy.sleep(1)
        assert RESULT.value == expected_value

    def test_send_director_scene(self):
        """
        Check stats handling of /director/scene messages.
        By sending this kind of message, trigger the stats message on /lg_stats/debug.

        """
        msg = GenericMessage(type="json", message="something")
        pub = rospy.Publisher("/director/scene", GenericMessage, queue_size=3)
        rospy.init_node(ROS_NODE_NAME, anonymous=True)
        self.checker(pub, msg, "something")

    def test_send_appctl_mode(self):
        """
        Check stats handling of /appctl/mode messages.
        By sending this kind of message, trigger the stats message on /lg_stats/debug.

        """
        msg = Mode(mode="a good mode")
        pub = rospy.Publisher("/appctl/mode", Mode, queue_size=3)
        rospy.init_node(ROS_NODE_NAME, anonymous=True)
        self.checker(pub, msg, "a good mode")

    def test_send_statistics_session(self):
        """
        Check stats handling of /statistics/session messages.
        By sending this kind of message, trigger the stats message on /lg_stats/debug.

        """
        msg = Session(application="a good application")
        pub = rospy.Publisher("/statistics/session", Session, queue_size=3)
        rospy.init_node(ROS_NODE_NAME, anonymous=True)
        self.checker(pub, msg, "a good application")

    def test_send_activity_active(self):
        """
        Check stats handling of /activity/active messages.
        By sending this kind of message, trigger the stats message on /lg_stats/debug.

        """
        msg = Bool(data=True)
        pub = rospy.Publisher("/activity/active", Bool, queue_size=3)
        rospy.init_node(ROS_NODE_NAME, anonymous=True)
        self.checker(pub, msg, "True")


if __name__ == "__main__":
    # pytest must provide result XML file just as rostest.rosrun would do
    # otherwise: FAILURE: test [test_lg_media_basic] did not generate test results
    test_pkg = ROS_NODE_NAME
    test_name = "test_lg_stats"
    test_dir = os.path.join(rospkg.get_test_results_dir(env=None), test_pkg)
    pytest_result_path = os.path.join(test_dir, "rosunit-%s.xml" % test_name)
    # run only itself
    test_path = os.path.abspath(os.path.abspath(__file__))
    # output is unfortunately handled / controlled by above layer of rostest (-s has no effect)
    pytest.main("%s -s -v --junit-xml=%s" % (test_path, pytest_result_path))
{'fraction_non_alphanumeric': 0.080608793686584, 'fraction_numerical': 0.0033821871476888386, 'mean_word_length': 2.6884466884466884, 'pattern_counts': {'":': 7, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 6}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3383009', 'n_tokens_mistral': 3731, 'n_tokens_neox': 3616, 'n_words': 1153}