fileexport_test.py

import pytz
import re
import subprocess
import unittest

from datetime import datetime
import predictionio
from predictionio import EventClient

app_name ='FileExporterApp'
access_key = 'FILE_EXPORT_TEST'
filename = 'export_events.json'

class FileExporterTest(unittest.TestCase):

    def setUp(self):
        subprocess.call(['pio', 'app', 'new', '--access-key', access_key, app_name])

    def tearDown(self):
        subprocess.call(['pio', 'app', 'delete', '-f', app_name])

    def test_export(self):
        app_info = subprocess.check_output(['pio', 'app', 'show', app_name])
        app_id = re.search('App ID: ([0-9]+)', app_info.decode('utf-8')).group(1)
        print(app_id)

        exporter = predictionio.FileExporter(file_name=filename)

        first_event_properties = {
            "prop1" : 1,
            "prop2" : "value2",
            "prop3" : [1, 2, 3],
            "prop4" : True,
            "prop5" : ["a", "b", "c"],
            "prop6" : 4.56 ,
            }
        first_event_time = datetime(
            2004, 12, 13, 21, 39, 45, 618000, pytz.timezone('US/Mountain'))
        exporter.create_event(
            event="my_event",
            entity_type="user",
            entity_id="uid",
            properties=first_event_properties,
            event_time=first_event_time,
            )

        # Second event
        second_event_properties = {
            "someProperty" : "value1",
            "anotherProperty" : "value2",
            }
        exporter.create_event(
            event="my_event",
            entity_type="user",
            entity_id="uid",
            target_entity_type="item",
            target_entity_id="iid",
            properties=second_event_properties,
            event_time=datetime(2014, 12, 13, 21, 38, 45, 618000, pytz.utc))

        exporter.close()

        subprocess.call(['pio', 'import', '--appid', app_id, '--input ', filename])

        # TODO
        # client = EventClient(access_key=access_key, url="http://127.0.0.1:7070")
        #
        # print("Get Event")
        # event = client.get_event(event_id)
        # print(event)
        # self.assertEqual(event.get('eventId'), event_id)

if __name__ == "__main__":
    unittest.main()