Merge pull request #1 from open-craft/dragonfi/add-various-utils

[WIP] Extract common functionality to xblock-utils

.gitignore

+*.pyc
+xblock_utils.egg-info

README.md

-xblock-utils
-============
+# xblock-utils: Various utilities for XBlocks
+
+These are a collection of useful utility functions,
+test base classes and documentation shared by many XBlocks.
+(Especially those of [edx-solutions](https://github.com/edx-solutions).)
+
+To test the utilities, run:
+
+    nosetests --with-coverage --cover-package xblockutils --cover-html

requirements.txt

+-e .
+
+# XBlock
+# This is not in/from PyPi, since it moves fast
+
+-e git+https://github.com/edx/XBlock.git@f0e53538be7ce90584a03cc7dd3f06bd43e12ac2#egg=XBlock
+

setup.py

+"""Set up for xblock-utils"""
+import os
+import os.path
+from setuptools import setup
+
+setup(
+    name='xblock-utils',
+    version='0.1a0',
+    description='Various utilities for XBlocks',
+    packages=[
+        'xblockutils',
+    ],
+    install_requires=[
+        'XBlock',
+    ]
+)

tests/integration/test_base_test.py

+import unittest
+
+from xblockutils.base_test import SeleniumBaseTest
+
+
+class TestSeleniumBaseTest(SeleniumBaseTest):
+    module_name = __name__
+    default_css_selector = "div.vertical"
+
+    def test_true(self):
+        self.go_to_page("Simple Scenario")
+
+
+class TestSeleniumBaseTestWithoutDefaultSelector(SeleniumBaseTest):
+    module_name = __name__
+
+    def test_true(self):
+        self.go_to_page("Simple Scenario", "div.vertical")

tests/integration/xml/simple_scenario.xml

+<vertical_demo>
+    <thumbs />
+    <thumbs />
+    <thumbs />
+</vertical_demo>

tests/unit/test_publish_event.py

+import unittest
+import json
+
+from xblockutils.publish_event import PublishEventMixin
+
+
+class EmptyMock():
+    pass
+
+
+class RequestMock(object):
+    method = "POST"
+
+    def __init__(self, data):
+        self.body = json.dumps(data)
+
+
+class RuntimeMock(object):
+    last_call = None
+
+    def publish(self, block, event_type, data):
+        self.last_call = (block, event_type, data)
+
+
+class XBlockMock(object):
+    def __init__(self):
+        self.runtime = RuntimeMock()
+
+
+class ObjectUnderTest(XBlockMock, PublishEventMixin):
+    pass
+
+
+class TestPublishEventMixin(unittest.TestCase):
+    def assert_no_calls_made(self, block):
+        self.assertFalse(block.last_call)
+
+    def assert_success(self, response):
+        self.assertEquals(json.loads(response.body)['result'], 'success')
+
+    def assert_error(self, response):
+        self.assertEquals(json.loads(response.body)['result'], 'error')
+
+    def test_error_when_no_event_type(self):
+        block = ObjectUnderTest()
+
+        response = block.publish_event(RequestMock({}))
+
+        self.assert_error(response)
+        self.assert_no_calls_made(block.runtime)
+
+    def test_uncustomized_publish_event(self):
+        block = ObjectUnderTest()
+
+        event_data = {"one": 1, "two": 2, "bool": True}
+        data = dict(event_data)
+        data["event_type"] = "test.event.uncustomized"
+
+        response = block.publish_event(RequestMock(data))
+
+        self.assert_success(response)
+        self.assertEquals(block.runtime.last_call, (block, "test.event.uncustomized", event_data))
+
+    def test_publish_event_with_additional_data(self):
+        block = ObjectUnderTest()
+        block.additional_publish_event_data = {"always_present": True, "block_id": "the-block-id"}
+
+        event_data = {"foo": True, "bar": False, "baz": None}
+        data = dict(event_data)
+        data["event_type"] = "test.event.customized"
+
+        response = block.publish_event(RequestMock(data))
+
+        expected_data = dict(event_data)
+        expected_data.update(block.additional_publish_event_data)
+
+        self.assert_success(response)
+        self.assertEquals(block.runtime.last_call, (block, "test.event.customized", expected_data))
+
+    def test_publish_event_fails_with_duplicate_data(self):
+        block = ObjectUnderTest()
+        block.additional_publish_event_data = {"good_argument": True, "clashing_argument": True}
+
+        event_data = {"fine_argument": True, "clashing_argument": False}
+        data = dict(event_data)
+        data["event_type"] = "test.event.clashing"
+
+        response = block.publish_event(RequestMock(data))
+
+        self.assert_error(response)
+        self.assert_no_calls_made(block.runtime)

xblockutils/base_test.py

+import os
+import sys
+import time
+
+import pkg_resources
+
+from django.template import Context, Template
+
+from selenium.webdriver.support.ui import WebDriverWait
+
+from workbench import scenarios
+from workbench.test.selenium_test import SeleniumTest
+
+
+class SeleniumBaseTest(SeleniumTest):
+    module_name = None
+    default_css_selector = None
+    relative_scenario_path = 'xml'
+    timeout = 10  # seconds
+
+    @property
+    def _module_name(self):
+        if self.module_name is None:
+            raise NotImplementedError("Overwrite cls.module_name in your derived class.")
+        return self.module_name
+
+    @property
+    def _default_css_selector(self):
+        if self.default_css_selector is None:
+            raise NotImplementedError("Overwrite cls.default_css_selector in your derived class.")
+        return self.default_css_selector
+
+    @property
+    def scenario_path(self):
+        base_dir = os.path.dirname(os.path.realpath(sys.modules[self._module_name].__file__))
+        return os.path.join(base_dir, self.relative_scenario_path)
+
+    def setUp(self):
+        super(SeleniumBaseTest, self).setUp()
+
+        # Use test scenarios
+        self.browser.get(self.live_server_url)  # Needed to load tests once
+        scenarios.SCENARIOS.clear()
+        scenarios_list = self._load_scenarios_from_path(self.scenario_path)
+        for identifier, title, xml in scenarios_list:
+            scenarios.add_xml_scenario(identifier, title, xml)
+            self.addCleanup(scenarios.remove_scenario, identifier)
+
+        # Suzy opens the browser to visit the workbench
+        self.browser.get(self.live_server_url)
+
+        # She knows it's the site by the header
+        header1 = self.browser.find_element_by_css_selector('h1')
+        self.assertEqual(header1.text, 'XBlock scenarios')
+
+    def wait_until_hidden(self, elem):
+        wait = WebDriverWait(elem, self.timeout)
+        wait.until(lambda e: not e.is_displayed(), u"{} should be hidden".format(elem.text))
+
+    def wait_until_disabled(self, elem):
+        wait = WebDriverWait(elem, self.timeout)
+        wait.until(lambda e: not e.is_enabled(), u"{} should be disabled".format(elem.text))
+
+    def wait_until_clickable(self, elem):
+        wait = WebDriverWait(elem, self.timeout)
+        wait.until(lambda e: e.is_displayed() and e.is_enabled(), u"{} should be cliclable".format(elem.text))
+
+    def wait_until_text_in(self, text, elem):
+        wait = WebDriverWait(elem, self.timeout)
+        wait.until(lambda e: text in e.text, u"{} should be in {}".format(text, elem.text))
+
+    def go_to_page(self, page_name, css_selector=None):
+        """
+        Navigate to the page `page_name`, as listed on the workbench home
+        Returns the DOM element on the visited page located by the `css_selector`
+        """
+        if css_selector is None:
+            css_selector = self._default_css_selector
+
+        self.browser.get(self.live_server_url)
+        self.browser.find_element_by_link_text(page_name).click()
+        time.sleep(1)
+        block = self.browser.find_element_by_css_selector(css_selector)
+        return block
+
+    def _load_resource(self, resource_path):
+        """
+        Gets the content of a resource
+        """
+        resource_content = pkg_resources.resource_string(self._module_name, resource_path)
+        return unicode(resource_content)
+
+    def _render_template(self, template_path, context={}):
+        """
+        Evaluate a template by resource path, applying the provided context
+        """
+        template_str = self._load_resource(template_path)
+        template = Template(template_str)
+        return template.render(Context(context))
+
+    def _load_scenarios_from_path(self, xml_path):
+        list_of_scenarios = []
+        if os.path.isdir(xml_path):
+            for template in os.listdir(xml_path):
+                if not template.endswith('.xml'):
+                    continue
+                identifier = template[:-4]
+                title = identifier.replace('_', ' ').title()
+                template_path = os.path.join(self.relative_scenario_path, template)
+                scenario = unicode(self._render_template(template_path, {"url_name": identifier}))
+                list_of_scenarios.append((identifier, title, scenario))
+        return list_of_scenarios

xblockutils/publish_event.py

+from xblock.core import XBlock
+
+
+class PublishEventMixin(object):
+    """A mixin for publishing events from an XBlock
+
+    Reuqires the object to have a runtime.publish method.
+    """
+    additional_publish_event_data = {}
+
+    @XBlock.json_handler
+    def publish_event(self, data, suffix=''):
+        try:
+            event_type = data.pop('event_type')
+        except KeyError as e:
+            return {'result': 'error', 'message': 'Missing event_type in JSON data'}
+
+        return self.publish_event_from_python(event_type, data)
+
+    def publish_event_from_python(self, event_type, data):
+        for key, value in self.additional_publish_event_data.items():
+            if key in data:
+                return {'result': 'error', 'message': 'Key should not be in publish_event data: {}'.format(key)}
+            data[key] = value
+
+        self.runtime.publish(self, event_type, data)
+        return {'result': 'success'}