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

PKG = 'lg_earth'
NAME = 'test_kmlsync'
KMLSYNC_HOST = '127.0.0.1'
KMLSYNC_PORT = 8765
KML_ENDPOINT = 'http://' + KMLSYNC_HOST + ':' + str(KMLSYNC_PORT)

WINDOW_SLUG = 'test_window_slug'

import sys
import re
import time
import json
import rospy
import rostest
import unittest
import requests
import xml.etree.ElementTree as ET
from std_msgs.msg import String
from xml.sax.saxutils import escape
from lg_common.helpers import escape_asset_url, generate_cookie
from interactivespaces_msgs.msg import GenericMessage
from subprocess import Popen
QUERY_TOPIC = '/earth/query/tour'
SCENE_TOPIC = '/director/scene'
LPNODE = 'testing_kmlsync_node'

DIRECTOR_MESSAGE = """
    {
            "description": "bogus",
            "duration": 0,
            "name": "test whatever",
            "resource_uri": "bogus",
            "slug": "test message",
            "windows": [
            {
                "activity": "earth",
                "assets": [
                        "http://lg-head:8060/media.kml",
                        "http://lg-head:8060/media/blah.kml",
                        "http://lg-head/zomgflolrlmao.kml"
                ],
                "height": 1080,
                "presentation_viewport": "center",
                "width": 1920,
                "x_coord": 0,
                "y_coord": 0
            },
            {
                "activity": "earth",
                "assets": [
                        "http://lg-head:8060/blah/right_one_content.kml"
                ],
                "height": 1080,
                "presentation_viewport": "right_one",
                "width": 1920,
                "x_coord": 0,
                "y_coord": 0
            }
            ]
    }
    """


class TestKMLSync(unittest.TestCase):
    def setUp(self):
        self.session = requests.Session()
        rospy.Subscriber(QUERY_TOPIC, String, self._listen_query_string)
        self.wait_for_http()
        self.query_string = ''

    def tearDown(self):
        self.session.close()

    def _listen_query_string(self, msg):
        self.query_string = msg.data

    def get_director_msg(self):
        msg = GenericMessage()
        msg.type = 'json'
        msg.message = DIRECTOR_MESSAGE
        return msg

    def get_request(self, url):
        r = self.session.get(url, timeout=0.1, stream=False)
        return r

    def wait_for_pubsub(self):
        # wait at most 5 seconds for listenerpublisher to be registered
        timeout_t = time.time() + 5.0
        while not rostest.is_subscriber(
            rospy.resolve_name(PUBTOPIC),
            rospy.resolve_name(LPNODE)) and time.time() < timeout_t:
            time.sleep(0.1)

        self.assert_(rostest.is_subscriber(
            rospy.resolve_name(PUBTOPIC),
            rospy.resolve_name(LPNODE)), "%s is not up"%LPNODE)

    def wait_for_http(self):
        # TODO: implement this
        rospy.sleep(3.0)

    def test_1_master_kml_200(self):
        r = self.get_request(KML_ENDPOINT + '/master.kml')
        result = r.status_code
        expected = 200
        self.assertEqual(result, expected)

    def test_2_master_kml_content(self):
        r = self.get_request(KML_ENDPOINT + '/master.kml')
        result = ET.fromstring(r.content).find('.//{http://www.opengis.net/kml/2.2}Document').attrib['id']
        expected = 'master'
        self.assertEqual(result, expected)

    def test_3_network_link_update_kml_without_params(self):
        r = self.get_request(KML_ENDPOINT + '/network_link_update.kml')
        result = r.status_code
        expected = 400
        self.assertEqual(result, expected)

    def test_4_network_link_update_cookie_string_is_initially_empty(self):
        r = self.get_request(KML_ENDPOINT + '/network_link_update.kml?window_slug=' + WINDOW_SLUG)
        result = get_cookie_string(r.content)
        expected = ''
        self.assertEqual(result, expected)

    def test_5_cookie_string_from_director(self):
        """
         - send director message
         - assert assets and cookie string<commit_msg>breaking test intentionally, touch: #31, #16<commit_after>#!/usr/bin/env python

PKG = 'lg_earth'
NAME = 'test_kmlsync'
KMLSYNC_HOST = '127.0.0.1'
KMLSYNC_PORT = 8765
KML_ENDPOINT = 'http://' + KMLSYNC_HOST + ':' + str(KMLSYNC_PORT)

WINDOW_SLUG = 'test_window_slug'

import sys
import re
import time
import json
import rospy
import rostest
import unittest
import requests
import xml.etree.ElementTree as ET
from std_msgs.msg import String
from xml.sax.saxutils import escape
from lg_common.helpers import escape_asset_url, generate_cookie
from interactivespaces_msgs.msg import GenericMessage
from subprocess import Popen
QUERY_TOPIC = '/earth/query/tour'
SCENE_TOPIC = '/director/scene'
LPNODE = 'testing_kmlsync_node'

DIRECTOR_MESSAGE = """
    {
            "description": "bogus",
            "duration": 0,
            "name": "test whatever",
            "resource_uri": "bogus",
            "slug": "test message",
            "windows": [
            {
                "activity": "earth",
                "assets": [
                        "http://lg-head:8060/media.kml",
                        "http://lg-head:8060/media/blah.kml",
                        "http://lg-head/zomgflolrlmao.kml"
                ],
                "height": 1080,
                "presentation_viewport": "center",
                "width": 1920,
                "x_coord": 0,
                "y_coord": 0
            },
            {
                "activity": "earth",
                "assets": [
                        "http://lg-head:8060/blah/right_one_content.kml"
                ],
                "height": 1080,
                "presentation_viewport": "right_one",
                "width": 1920,
                "x_coord": 0,
                "y_coord": 0
            }
            ]
    }
    """


class TestKMLSync(unittest.TestCase):
    def setUp(self):
        self.session = requests.Session()
        rospy.Subscriber(QUERY_TOPIC, String, self._listen_query_string)
        self.wait_for_http()
        self.query_string = ''

    def tearDown(self):
        self.session.close()

    def _listen_query_string(self, msg):
        self.query_string = msg.data

    def get_director_msg(self):
        msg = GenericMessage()
        msg.type = 'json'
        msg.message = DIRECTOR_MESSAGE
        return msg

    def get_request(self, url):
        r = self.session.get(url, timeout=0.1, stream=False)
        return r

    def wait_for_pubsub(self):
        # wait at most 5 seconds for listenerpublisher to be registered
        timeout_t = time.time() + 5.0
        while not rostest.is_subscriber(
            rospy.resolve_name(PUBTOPIC),
            rospy.resolve_name(LPNODE)) and time.time() < timeout_t:
            time.sleep(0.1)

        self.assert_(rostest.is_subscriber(
            rospy.resolve_name(PUBTOPIC),
            rospy.resolve_name(LPNODE)), "%s is not up"%LPNODE)

    def wait_for_http(self):
        # TODO: implement this
        rospy.sleep(3.0)

    def test_1_master_kml_200(self):
        r = self.get_request(KML_ENDPOINT + '/master.kml')
        result = r.status_code
        expected = 200
        self.assertEqual(result, expected)

    def test_2_master_kml_content(self):
        r = self.get_request(KML_ENDPOINT + '/master.kml')
        result = ET.fromstring(r.content).find('.//{http://www.opengis.net/kml/2.2}Document').attrib['id']
        expected = 'master'
        self.assertEqual(result, expected)

    def test_3_network_link_update_kml_without_params(self):
        r = self.get_request(KML_ENDPOINT + '/network_link_update.kml')
        result = r.status_code
        expected = 400
        self.assertEqual(result, expected)
        self.assertEqual(1, 2)

    def test_4_network_link_update_cookie_string_is_initially_empty(self):
        r = self.get_request(KML_ENDPOINT + '/network_link_update.kml?window_slug=' + WINDOW_SLUG)
        result = get_cookie_string(r.content)
        expected = ''
        self.assertEqual(result, expected)

    def test_5_cookie_string_from_director(self):
        """
         - send director message
         - assert assets and cookie string
{'fraction_non_alphanumeric': 0.11088170998299733, 'fraction_numerical': 0.017245567160553803, 'mean_word_length': 2.0900562851782363, 'pattern_counts': {'":': 40, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 4}, 'pii_count': 2, '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': '394421', 'n_tokens_mistral': 2731, 'n_tokens_neox': 2570, 'n_words': 478}