This repository has been archived by the owner on Oct 9, 2018. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 1
/
PythonConduitServerTest.cpp
105 lines (70 loc) · 3.35 KB
/
PythonConduitServerTest.cpp
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
/*
* ConduitServerTest.cpp
* MonkeyWorksCore
*
* Created by David Cox on 12/8/09.
* Copyright 2009 Harvard University. All rights reserved.
*
*/
#include "PythonConduitServerTest.h"
/*
* EventStreamConduitTest.cpp
* MonkeyWorksCore
*
* Created by David Cox on 11/12/09.
* Copyright 2009 Harvard University. All rights reserved.
*
*/
#include "DummyEventTransport.h"
#include "IPCEventTransport.h"
#include "CodecAwareConduit.h"
#include "StandardSystemEventHandler.h"
#include "StandardVariables.h"
#include "EventStreamConduit.h"
using namespace mw;
//CPPUNIT_TEST_SUITE_NAMED_REGISTRATION( PythonConduitServerTestFixture, "Unit Test" );
void PythonConduitServerTestFixture::setUp(){
FullCoreEnvironmentTestFixture::setUp();
event_handler = shared_ptr<EventStreamInterface>(new StandardSystemEventHandler());
}
void PythonConduitServerTestFixture::testInOneThread(){
// Create Interprocess event transports. In principle, these could be
// in other processes
// NOTE: the resource name must agree with that in PythonConduitTest.py
shared_ptr<EventTransport> serverside_transport(new IPCEventTransport(EventTransport::server_event_transport,
EventTransport::bidirectional_event_transport,
"python_test_conduit"));
// Create the conduits to test
shared_ptr<CodecAwareConduit> server_conduit(new CodecAwareConduit(serverside_transport));
server_conduit->initialize();
// Launch the python process
// need to figure out what to do with the paths
execl("/usr/bin/python", "../../PythonConduitTest.py");
// the following values must agree with those in PythonConduitTest.py
int code_to_use = 4;
string var_to_use = ANNOUNCE_BLOCK_TAGNAME;
// Create some dummy event handlers. The collector objects just catch
// events and save them for inspection
shared_ptr<SimpleEventCollector> serverside_collector_a(new SimpleEventCollector());
shared_ptr<SimpleEventCollector> serverside_collector_b(new SimpleEventCollector());
server_conduit->registerCallbackByName(ANNOUNCE_BLOCK_TAGNAME, bind(&SimpleEventCollector::handleEvent, serverside_collector_a, _1));
// Start the conduits runnings
shared_ptr<Clock> clock = Clock::instance();
clock->sleepMS(100);
shared_ptr<Variable> a_standard_variable = global_variable_registry->getVariable(var_to_use);
a_standard_variable->setValue(4.0);
a_standard_variable->setValue(5.0);
a_standard_variable->setValue(6.0);
server_conduit->sendData(code_to_use, 6L);
server_conduit->sendData(code_to_use, "blah");
server_conduit->sendData(code_to_use, 8.0);
clock->sleepMS(100);
shared_ptr<Event> received_event_a = serverside_collector_a->getLastEvent();
shared_ptr<Event> received_event_b = serverside_collector_b->getLastEvent();
CPPUNIT_ASSERT(received_event_a != NULL);
CPPUNIT_ASSERT(received_event_a->getEventCode() == 200);
CPPUNIT_ASSERT((double)received_event_a->getData() == (double)3.0);
CPPUNIT_ASSERT(received_event_b != NULL);
CPPUNIT_ASSERT(received_event_b->getEventCode() == 201);
CPPUNIT_ASSERT((double)received_event_b->getData() == (double)6.0);
}