|
84 | 84 | unregister_counter.incrementAndGet |
85 | 85 | completable_future |
86 | 86 | } |
87 | | - expect(EventProcessorHost).to receive(:new).at_most(2).times {|host_name, event_hub_name, consumer_group, event_hub_connection, storage_connection, container, executor| |
88 | | - case event_hub_name |
89 | | - when 'event_hub_name0' |
90 | 87 |
|
91 | | - assertion_count.incrementAndGet |
| 88 | + build_step_mock = double("final build step") |
| 89 | + expect(build_step_mock).to receive(:build).at_most(2).times.and_return(mock_host) |
| 90 | + |
| 91 | + executor_step = double("executor step") |
| 92 | + expect(executor_step).to receive(:setExecutor).at_most(2).times.and_return(build_step_mock) |
| 93 | + |
| 94 | + mock_connection_string = double("connection string") |
| 95 | + expect(mock_connection_string).to receive(:useEventHubConnectionString).at_most(2).times {|event_hub_connection| |
| 96 | + case event_hub_connection |
| 97 | + when /.*event_hub_name0$/ |
92 | 98 | expect(event_hub_connection).to eql(config['event_hub_connections'][0]) |
93 | | - expect(container).to eql('event_hub_name0') # default |
| 99 | + when /.*event_hub_name1$/ |
| 100 | + expect(event_hub_connection).to eql(config['event_hub_connections'][1]) |
| 101 | + end |
| 102 | + executor_step |
| 103 | + } |
94 | 104 |
|
| 105 | + mock_builder = double("storage and lease managers") |
| 106 | + expect(mock_builder).to receive(:useAzureStorageCheckpointLeaseManager).at_most(2).times {|storage_connection_str, storage_container, storage_blob_prefix| |
| 107 | + case storage_container |
| 108 | + when 'event_hub_name0' |
| 109 | + assertion_count.incrementAndGet |
95 | 110 | when 'event_hub_name1' |
96 | 111 | assertion_count.incrementAndGet |
97 | | - expect(host_name).to start_with('logstash') |
98 | | - expect(event_hub_connection).to eql(config['event_hub_connections'][1]) |
99 | | - expect(container).to eql('event_hub_name1') # default |
100 | 112 | end |
101 | | - expect(host_name).to start_with('logstash') |
102 | | - expect(storage_connection).to eql(config['storage_connection']) |
| 113 | + mock_connection_string |
| 114 | + } |
103 | 115 |
|
| 116 | + expect(EventProcessorHost::EventProcessorHostBuilder).to receive(:newBuilder).at_most(2).times {|host_name, consumer_group| |
| 117 | + expect(host_name).to start_with('logstash') |
104 | 118 | host_counter.incrementAndGet |
105 | | - mock_host |
| 119 | + mock_builder |
106 | 120 | } |
107 | 121 | # signal the stop first since the run method blocks until stop is called. |
108 | 122 | input.do_stop |
|
126 | 140 | expect(exploded_config[0]['event_hub_connections'][0].value).to eql('Endpoint=sb://logstash/;SharedAccessKeyName=activity-log-readonly;SharedAccessKey=something;EntityPath=event_hub1') |
127 | 141 | end |
128 | 142 | end |
129 | | - |
130 | | - |
131 | 143 | end |
132 | 144 |
|
133 | 145 | describe "Advanced Config" do |
|
221 | 233 | unregister_counter.incrementAndGet |
222 | 234 | completable_future |
223 | 235 | } |
224 | | - expect(EventProcessorHost).to receive(:new).at_most(3).times {|host_name, event_hub_name, consumer_group, event_hub_connection, storage_connection, container, executor| |
225 | | - case event_hub_name |
| 236 | + |
| 237 | + build_step_mock = double("final build step") |
| 238 | + expect(build_step_mock).to receive(:build).at_most(3).times.and_return(mock_host) |
| 239 | + |
| 240 | + executor_step = double("executor step") |
| 241 | + expect(executor_step).to receive(:setExecutor).at_most(3).times.and_return(build_step_mock) |
| 242 | + |
| 243 | + mock_connection_string = double("connection string") |
| 244 | + expect(mock_connection_string).to receive(:useEventHubConnectionString).at_most(3).times {|event_hub_connection| |
| 245 | + case event_hub_connection |
| 246 | + when /.*event_hub_name0$/ |
| 247 | + expect(event_hub_connection).to eql(config['event_hubs'][0]['event_hub_name0']['event_hub_connections'][0].value) |
| 248 | + when /.*event_hub_name1$/ |
| 249 | + expect(event_hub_connection).to eql(config['event_hubs'][1]['event_hub_name1']['event_hub_connections'][0].value) |
| 250 | + end |
| 251 | + executor_step |
| 252 | + } |
| 253 | + |
| 254 | + managers_mock = double("checkpoint and lease managers") |
| 255 | + expect(managers_mock).to receive(:useUserCheckpointAndLeaseManagers).at_most(3).times {|checkpoint_mngr, lease_mngr| |
| 256 | + assertion_count.incrementAndGet |
| 257 | + mock_connection_string |
| 258 | + } |
| 259 | + expect(managers_mock).to receive(:useAzureStorageCheckpointLeaseManager).at_most(3).times {|storage_connection_str, storage_container, storage_blob_prefix| |
| 260 | + case storage_container |
226 | 261 | when 'event_hub_name0' |
227 | | - if consumer_group.eql?('cg') |
228 | | - assertion_count.incrementAndGet |
229 | | - expect(host_name).to start_with('logstash') |
230 | | - expect(event_hub_connection).to eql(config['event_hubs'][0]['event_hub_name0']['event_hub_connections'][0].value) |
231 | | - expect(storage_connection).to eql(config['event_hubs'][0]['event_hub_name0']['storage_connection'].value) |
232 | | - expect(container).to eql('event_hub_name0') # default |
233 | | - elsif consumer_group.eql?('ls') |
234 | | - assertion_count.incrementAndGet |
235 | | - expect(event_hub_connection).to eql(config['event_hubs'][2]['event_hub_name0']['event_hub_connections'][0].value) |
236 | | - # in this mode, storage connection and container are replaced with in memory offset management |
237 | | - expect(storage_connection).to be_kind_of(InMemoryCheckpointManager) |
238 | | - expect(container).to be_kind_of(InMemoryLeaseManager) |
239 | | - end |
240 | | - when 'event_hub_name1' |
241 | 262 | assertion_count.incrementAndGet |
242 | | - expect(host_name).to start_with('logstash') |
243 | | - expect(event_hub_connection).to eql(config['event_hubs'][1]['event_hub_name1']['event_hub_connections'][0].value) |
244 | | - expect(storage_connection).to eql(config['event_hubs'][1]['event_hub_name1']['storage_connection'].value) |
245 | | - expect(container).to eql(config['event_hubs'][1]['event_hub_name1']['storage_container']) |
| 263 | + when 'alt_container' |
| 264 | + assertion_count.incrementAndGet |
246 | 265 | end |
| 266 | + mock_connection_string |
| 267 | + } |
| 268 | + |
| 269 | + expect(EventProcessorHost::EventProcessorHostBuilder).to receive(:newBuilder).at_most(3).times {|host_name, consumer_group| |
| 270 | + expect(host_name).to start_with('logstash') |
247 | 271 | host_counter.incrementAndGet |
248 | | - mock_host |
| 272 | + managers_mock |
249 | 273 | } |
| 274 | + |
250 | 275 | # signal the stop first since the run method blocks until stop is called. |
251 | 276 | input.do_stop |
252 | 277 | input.run(mock_queue) |
|
259 | 284 | it "can create an in memory EPH" do |
260 | 285 | #event_hub, event_hub_name, scheduled_executor_service |
261 | 286 | exploded_config = input.event_hubs_exploded |
| 287 | + # During build step Azure libraries does a syntax check of EventHub connection string, so needs to be a pseudo real |
| 288 | + exploded_config[0]['event_hub_connections'] = [::LogStash::Util::Password.new("Endpoint=sb://logstash.windows.net/;SharedAccessKeyName=activity-log-read-only;SharedAccessKey=blabla;EntityPath=ops-logs")] |
262 | 289 | input.create_in_memory_event_processor_host(exploded_config[0], exploded_config[0]['event_hubs'].first, nil) |
263 | 290 | end |
264 | 291 | end |
265 | 292 |
|
266 | 293 | describe "Bad Basic Config" do |
267 | | - describe "Offset overwritting" do |
| 294 | + describe "Offset overwriting" do |
268 | 295 | let(:config) do |
269 | 296 | { |
270 | 297 | 'event_hub_connections' => ['Endpoint=sb://...;EntityPath=event_hub_name0', 'Endpoint=sb://...;EntityPath=event_hub_name0'], |
|
0 commit comments