Java 类org.easymock.Capture 实例源码

/** write a list of messages */
@Test(timeout = 5000)
public void testMessageWriteList() throws InterruptedException, ExecutionException {
    Capture<List<OFMessage>> cMsgList = prepareChannelForWriteList();

    OFHello hello = factory.hello(ImmutableList.<OFHelloElem>of());
    OFPacketOut packetOut = factory.buildPacketOut()
            .setData(new byte[] { 0x01, 0x02, 0x03, 0x04 })
            .setActions(ImmutableList.<OFAction>of( factory.actions().output(OFPort.of(1), 0)))
            .build();

    conn.write(ImmutableList.of(hello, packetOut));

    assertThat("Write should have been written", cMsgList.hasCaptured(), equalTo(true));
    List<OFMessage> value = cMsgList.getValue();
    logger.info("Captured channel write: "+value);
    assertThat("Should have captured MsgList", cMsgList.getValue(),
            Matchers.<OFMessage> contains(hello, packetOut));
}

/** write a list of messages */
@Test(timeout = 5000)
public void testMessageWriteList() throws InterruptedException, ExecutionException {
    Capture<List<OFMessage>> cMsgList = prepareChannelForWriteList();

    OFHello hello = factory.hello(ImmutableList.<OFHelloElem>of());
    OFPacketOut packetOut = factory.buildPacketOut()
            .setData(new byte[] { 0x01, 0x02, 0x03, 0x04 })
            .setActions(ImmutableList.<OFAction>of( factory.actions().output(OFPort.of(1), 0)))
            .build();

    conn.write(ImmutableList.of(hello, packetOut));
    eventLoop.runTasks();
    assertThat("Write should have been written", cMsgList.hasCaptured(), equalTo(true));
    List<OFMessage> value = cMsgList.getValue();
    logger.info("Captured channel write: "+value);
    assertThat("Should have captured MsgList", cMsgList.getValue(),
            Matchers.<OFMessage> contains(hello, packetOut));
}

@Test
public void shouldAcceptDefaultTimeWindow() {
    final TwoWayChannelMessage message = createClientVersionMessage();
    final Capture<TwoWayChannelMessage> initiateCapture = new Capture<>();
    connection.sendToClient(capture(initiateCapture));
    replay(connection);

    dut = new PaymentChannelServer(broadcaster, wallet, Coin.CENT, connection);

    dut.connectionOpen();
    dut.receiveMessage(message);

    long expectedExpire = Utils.currentTimeSeconds() + 24 * 60 * 60 - 60;  // This the default defined in paymentchannel.proto
    assertServerVersion();
    assertExpireTime(expectedExpire, initiateCapture);
}

@Test
public void shouldAllowExactTimeWindow() {
    final TwoWayChannelMessage message = createClientVersionMessage();
    final Capture<TwoWayChannelMessage> initiateCapture = new Capture<>();
    connection.sendToClient(capture(initiateCapture));
    replay(connection);
    final int expire = 24 * 60 * 60 - 60;  // This the default defined in paymentchannel.proto

    dut = new PaymentChannelServer(broadcaster, wallet, Coin.CENT, new PaymentChannelServer.DefaultServerChannelProperties(){
        @Override
        public long getMaxTimeWindow() { return expire; }
        @Override
        public long getMinTimeWindow() { return expire; }
    }, connection);
    dut.connectionOpen();
    long expectedExpire = Utils.currentTimeSeconds() + expire;
    dut.receiveMessage(message);

    assertServerVersion();
    assertExpireTime(expectedExpire, initiateCapture);
}

@Test
public final void testUpdatePortStatus() {
    putDevice(DID1, SW1);
    List<PortDescription> pds = Arrays.<PortDescription>asList(
            new DefaultPortDescription(P1, true)
            );
    deviceStore.updatePorts(PID, DID1, pds);

    Capture<InternalPortStatusEvent> message = new Capture<>();
    Capture<MessageSubject> subject = new Capture<>();
    Capture<Function<InternalPortStatusEvent, byte[]>> encoder = new Capture<>();

    resetCommunicatorExpectingSingleBroadcast(message, subject, encoder);
    final DefaultPortDescription desc = new DefaultPortDescription(P1, false);
    DeviceEvent event = deviceStore.updatePortStatus(PID, DID1, desc);
    assertEquals(PORT_UPDATED, event.type());
    assertDevice(DID1, SW1, event.subject());
    assertEquals(P1, event.port().number());
    assertFalse("Port is disabled", event.port().isEnabled());
    verify(clusterCommunicator);
    assertInternalPortStatusEvent(NID1, DID1, PID, desc, NO_ANNOTATION, message, subject, encoder);
    assertTrue(message.hasCaptured());
}

@Test
public void testSendRecordsConvertsData() throws Exception {
    createWorkerTask();

    List<SourceRecord> records = new ArrayList<>();
    // Can just use the same record for key and value
    records.add(new SourceRecord(PARTITION, OFFSET, "topic", null, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD));

    Capture<ProducerRecord<byte[], byte[]>> sent = expectSendRecordAnyTimes();

    PowerMock.replayAll();

    Whitebox.setInternalState(workerTask, "toSend", records);
    Whitebox.invokeMethod(workerTask, "sendRecords");
    assertEquals(SERIALIZED_KEY, sent.getValue().key());
    assertEquals(SERIALIZED_RECORD, sent.getValue().value());

    PowerMock.verifyAll();
}

@Test
public void testSendRecordsPropagatesTimestamp() throws Exception {
    final Long timestamp = System.currentTimeMillis();

    createWorkerTask();

    List<SourceRecord> records = Collections.singletonList(
            new SourceRecord(PARTITION, OFFSET, "topic", null, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD, timestamp)
    );

    Capture<ProducerRecord<byte[], byte[]>> sent = expectSendRecordAnyTimes();

    PowerMock.replayAll();

    Whitebox.setInternalState(workerTask, "toSend", records);
    Whitebox.invokeMethod(workerTask, "sendRecords");
    assertEquals(timestamp, sent.getValue().timestamp());

    PowerMock.verifyAll();
}

@Test(expected = InvalidRecordException.class)
public void testSendRecordsCorruptTimestamp() throws Exception {
    final Long timestamp = -3L;
    createWorkerTask();

    List<SourceRecord> records = Collections.singletonList(
            new SourceRecord(PARTITION, OFFSET, "topic", null, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD, timestamp)
    );

    Capture<ProducerRecord<byte[], byte[]>> sent = expectSendRecordAnyTimes();

    PowerMock.replayAll();

    Whitebox.setInternalState(workerTask, "toSend", records);
    Whitebox.invokeMethod(workerTask, "sendRecords");
    assertEquals(null, sent.getValue().timestamp());

    PowerMock.verifyAll();
}

@Test
public void testSendRecordsNoTimestamp() throws Exception {
    final Long timestamp = -1L;
    createWorkerTask();

    List<SourceRecord> records = Collections.singletonList(
            new SourceRecord(PARTITION, OFFSET, "topic", null, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD, timestamp)
    );

    Capture<ProducerRecord<byte[], byte[]>> sent = expectSendRecordAnyTimes();

    PowerMock.replayAll();

    Whitebox.setInternalState(workerTask, "toSend", records);
    Whitebox.invokeMethod(workerTask, "sendRecords");
    assertEquals(null, sent.getValue().timestamp());

    PowerMock.verifyAll();
}

private void expectApplyTransformationChain(boolean anyTimes) {
    final Capture<SourceRecord> recordCapture = EasyMock.newCapture();
    IExpectationSetters<SourceRecord> convertKeyExpect = EasyMock.expect(transformationChain.apply(EasyMock.capture(recordCapture)));
    if (anyTimes)
        convertKeyExpect.andStubAnswer(new IAnswer<SourceRecord>() {
            @Override
            public SourceRecord answer() {
                return recordCapture.getValue();
            }
        });
    else
        convertKeyExpect.andAnswer(new IAnswer<SourceRecord>() {
            @Override
            public SourceRecord answer() {
                return recordCapture.getValue();
            }
        });
}

@Test
public void testSchedule() throws Exception {
    Capture<Runnable> taskWrapper = EasyMock.newCapture();

    ScheduledFuture commitFuture = PowerMock.createMock(ScheduledFuture.class);
    EasyMock.expect(executor.scheduleWithFixedDelay(
            EasyMock.capture(taskWrapper), eq(DEFAULT_OFFSET_COMMIT_INTERVAL_MS),
            eq(DEFAULT_OFFSET_COMMIT_INTERVAL_MS), eq(TimeUnit.MILLISECONDS))
    ).andReturn(commitFuture);

    ConnectorTaskId taskId = PowerMock.createMock(ConnectorTaskId.class);
    WorkerSourceTask task = PowerMock.createMock(WorkerSourceTask.class);

    EasyMock.expect(committers.put(taskId, commitFuture)).andReturn(null);

    PowerMock.replayAll();

    committer.schedule(taskId, task);
    assertTrue(taskWrapper.hasCaptured());
    assertNotNull(taskWrapper.getValue());

    PowerMock.verifyAll();
}

private void assertInternalDeviceEvent(NodeId sender,
        DeviceId deviceId,
        ProviderId providerId,
        DeviceDescription expectedDesc,
        Capture<InternalDeviceEvent> actualEvent,
        Capture<MessageSubject> actualSubject,
        Capture<Function<InternalDeviceEvent, byte[]>> actualEncoder) {
    assertTrue(actualEvent.hasCaptured());
    assertTrue(actualSubject.hasCaptured());
    assertTrue(actualEncoder.hasCaptured());

    assertEquals(GossipDeviceStoreMessageSubjects.DEVICE_UPDATE,
            actualSubject.getValue());
    assertEquals(deviceId, actualEvent.getValue().deviceId());
    assertEquals(providerId, actualEvent.getValue().providerId());
    assertDeviceDescriptionEquals(expectedDesc, actualEvent.getValue().deviceDescription().value());
}

/** write a packetOut, which is buffered */
@Test(timeout = 5000)
public void testSingleMessageWrite() throws InterruptedException, ExecutionException {
    Capture<List<OFMessage>> cMsgList = prepareChannelForWriteList();

    OFPacketOut packetOut = factory.buildPacketOut()
            .setData(new byte[] { 0x01, 0x02, 0x03, 0x04 })
            .setActions(ImmutableList.<OFAction>of( factory.actions().output(OFPort.of(1), 0)))
            .build();

    conn.write(packetOut);
    assertThat("Write should have been flushed", cMsgList.hasCaptured(), equalTo(true));

    List<OFMessage> value = cMsgList.getValue();
    logger.info("Captured channel write: "+value);
    assertThat("Should have captured MsgList", cMsgList.getValue(),
            Matchers.<OFMessage> contains(packetOut));
}

private void expectConvertWriteRead(final String configKey, final Schema valueSchema, final byte[] serialized,
                                    final String dataFieldName, final Object dataFieldValue) {
    final Capture<Struct> capturedRecord = EasyMock.newCapture();
    if (serialized != null)
        EasyMock.expect(converter.fromConnectData(EasyMock.eq(TOPIC), EasyMock.eq(valueSchema), EasyMock.capture(capturedRecord)))
                .andReturn(serialized);
    storeLog.send(EasyMock.eq(configKey), EasyMock.aryEq(serialized));
    PowerMock.expectLastCall();
    EasyMock.expect(converter.toConnectData(EasyMock.eq(TOPIC), EasyMock.aryEq(serialized)))
            .andAnswer(new IAnswer<SchemaAndValue>() {
                @Override
                public SchemaAndValue answer() throws Throwable {
                    if (dataFieldName != null)
                        assertEquals(dataFieldValue, capturedRecord.getValue().get(dataFieldName));
                    // Note null schema because default settings for internal serialization are schema-less
                    return new SchemaAndValue(null, serialized == null ? null : structToMap(capturedRecord.getValue()));
                }
            });
}

@Test
public void testRestartConnectorLeaderRedirect() throws Throwable {
    final Capture<Callback<Void>> cb = Capture.newInstance();
    herder.restartConnector(EasyMock.eq(CONNECTOR_NAME), EasyMock.capture(cb));
    expectAndCallbackNotLeaderException(cb);

    EasyMock.expect(RestServer.httpRequest(EasyMock.eq("http://leader:8083/connectors/" + CONNECTOR_NAME + "/restart?forward=true"),
            EasyMock.eq("POST"), EasyMock.isNull(), EasyMock.<TypeReference>anyObject()))
            .andReturn(new RestServer.HttpResponse<>(202, new HashMap<String, List<String>>(), null));

    PowerMock.replayAll();

    connectorsResource.restartConnector(CONNECTOR_NAME, null);

    PowerMock.verifyAll();
}

@Test
public void testRestartConnectorOwnerRedirect() throws Throwable {
    final Capture<Callback<Void>> cb = Capture.newInstance();
    herder.restartConnector(EasyMock.eq(CONNECTOR_NAME), EasyMock.capture(cb));
    String ownerUrl = "http://owner:8083";
    expectAndCallbackException(cb, new NotAssignedException("not owner test", ownerUrl));

    EasyMock.expect(RestServer.httpRequest(EasyMock.eq("http://owner:8083/connectors/" + CONNECTOR_NAME + "/restart?forward=false"),
            EasyMock.eq("POST"), EasyMock.isNull(), EasyMock.<TypeReference>anyObject()))
            .andReturn(new RestServer.HttpResponse<>(202, new HashMap<String, List<String>>(), null));

    PowerMock.replayAll();

    connectorsResource.restartConnector(CONNECTOR_NAME, true);

    PowerMock.verifyAll();
}

@Test
public final void testMarkOffline() {

    putDevice(DID1, SW1);
    assertTrue(deviceStore.isAvailable(DID1));

    Capture<InternalDeviceEvent> message = new Capture<>();
    Capture<MessageSubject> subject = new Capture<>();
    Capture<Function<InternalDeviceEvent, byte[]>> encoder = new Capture<>();

    resetCommunicatorExpectingSingleBroadcast(message, subject, encoder);
    DeviceEvent event = deviceStore.markOffline(DID1);
    assertEquals(DEVICE_AVAILABILITY_CHANGED, event.type());
    assertDevice(DID1, SW1, event.subject());
    assertFalse(deviceStore.isAvailable(DID1));
    verify(clusterCommunicator);
    // TODO: verify broadcast message
    assertTrue(message.hasCaptured());


    resetCommunicatorExpectingNoBroadcast(message, subject, encoder);
    DeviceEvent event2 = deviceStore.markOffline(DID1);
    assertNull("No change, no event", event2);
    verify(clusterCommunicator);
    assertFalse(message.hasCaptured());
}

@Test(timeout = 5000)
public void testWriteRequestSuccess() throws InterruptedException, ExecutionException {
    Capture<List<OFMessage>> cMsgList = prepareChannelForWriteList();

    OFEchoRequest echoRequest = factory.echoRequest(new byte[] {});
    ListenableFuture<OFEchoReply> future = conn.writeRequest(echoRequest);
    assertThat("Connection should have 1 pending request",
            conn.getPendingRequestIds().size(), equalTo(1));

    assertThat("Should have captured MsgList", cMsgList.getValue(),
            Matchers.<OFMessage> contains(echoRequest));

    assertThat("Future should not be complete yet", future.isDone(), equalTo(false));

    OFEchoReply echoReply = factory.buildEchoReply()
            .setXid(echoRequest.getXid())
            .build();

    assertThat("Connection should have accepted the response",
            conn.deliverResponse(echoReply),
            equalTo(true));
    assertThat("Future should be complete ", future.isDone(), equalTo(true));
    assertThat(future.get(), equalTo(echoReply));
    assertThat("Connection should have no pending requests",
            conn.getPendingRequestIds().isEmpty(), equalTo(true));
}

@Test
public void putSafeWithNoPreviousValueIsPropagated() {
    final Converter converter = mock(Converter.class);
    final KafkaBasedLog<String, byte[]> kafkaBasedLog = mock(KafkaBasedLog.class);
    final KafkaStatusBackingStore store = new KafkaStatusBackingStore(new MockTime(), converter, STATUS_TOPIC, kafkaBasedLog);

    final byte[] value = new byte[0];

    final Capture<Struct> statusValueStruct = newCapture();
    converter.fromConnectData(eq(STATUS_TOPIC), anyObject(Schema.class), capture(statusValueStruct));
    EasyMock.expectLastCall().andReturn(value);

    kafkaBasedLog.send(eq("status-connector-" + CONNECTOR), eq(value), anyObject(Callback.class));
    expectLastCall();

    replayAll();

    final ConnectorStatus status = new ConnectorStatus(CONNECTOR, ConnectorStatus.State.FAILED, WORKER_ID, 0);
    store.putSafe(status);

    verifyAll();

    assertEquals(status.state().toString(), statusValueStruct.getValue().get(KafkaStatusBackingStore.STATE_KEY_NAME));
    assertEquals(status.workerId(), statusValueStruct.getValue().get(KafkaStatusBackingStore.WORKER_ID_KEY_NAME));
    assertEquals(status.generation(), statusValueStruct.getValue().get(KafkaStatusBackingStore.GENERATION_KEY_NAME));
}

private void assertInternalDeviceEvent(NodeId sender,
        DeviceId deviceId,
        ProviderId providerId,
        DeviceDescription expectedDesc,
        List<SparseAnnotations> expectedAnnotations,
        Capture<InternalDeviceEvent> actualEvent,
        Capture<MessageSubject> actualSubject,
        Capture<Function<InternalDeviceEvent, byte[]>> actualEncoder) {
    assertTrue(actualEvent.hasCaptured());
    assertTrue(actualSubject.hasCaptured());
    assertTrue(actualEncoder.hasCaptured());

    assertEquals(GossipDeviceStoreMessageSubjects.DEVICE_UPDATE,
            actualSubject.getValue());
    assertEquals(deviceId, actualEvent.getValue().deviceId());
    assertEquals(providerId, actualEvent.getValue().providerId());
    assertDeviceDescriptionEquals(
            expectedDesc,
            expectedAnnotations,
            actualEvent.getValue().deviceDescription().value());
}

private <T> void resetCommunicatorExpectingSingleBroadcast(
        Capture<T> message,
        Capture<MessageSubject> subject,
        Capture<Function<T, byte[]>> encoder) {

    message.reset();
    subject.reset();
    encoder.reset();
    reset(clusterCommunicator);
    clusterCommunicator.broadcast(
                capture(message),
                capture(subject),
                capture(encoder));
    expectLastCall().once();
    replay(clusterCommunicator);
}

@Test
public void installIngressFlowReplaceAction() throws Exception {
    Capture<OFFlowMod> capture = prepareForInstallTest();

    switchManager.installIngressFlow(dpid, cookieHex, cookie,
            inputPort, outputPort, inputVlanId, transitVlanId, OutputVlanType.REPLACE, meterId);

    assertEquals(
            scheme.ingressReplaceFlowMod(inputPort, outputPort, inputVlanId, transitVlanId, meterId, cookie),
            capture.getValue());
}

@Test
public void installIngressFlowPopAction() throws Exception {
    Capture<OFFlowMod> capture = prepareForInstallTest();

    switchManager.installIngressFlow(dpid, cookieHex, cookie,
            inputPort, outputPort, inputVlanId, transitVlanId, OutputVlanType.POP, meterId);

    assertEquals(
            scheme.ingressPopFlowMod(inputPort, outputPort, inputVlanId, transitVlanId, meterId, cookie),
            capture.getValue());
}

void mockLayerGroupList(List<LayerGroupInfo> groups) {
    Capture<Filter> filterCapture = new Capture<>();
    EasyMock.expect(catalog.list(EasyMock.eq(LayerGroupInfo.class), EasyMock.capture(filterCapture)))
        .andStubAnswer(()->{
            List<LayerGroupInfo> matchingGroups = groups.stream()
                .filter(x->filterCapture.getValue().evaluate(x))
                .collect(Collectors.toList());
            return new CloseableIteratorAdapter<LayerGroupInfo>(
                matchingGroups
                    .iterator()
                    );
            });
}

@Test
public void installIngressFlowNoneAction() throws Exception {
    Capture<OFFlowMod> capture = prepareForInstallTest();

    switchManager.installIngressFlow(dpid, cookieHex, cookie,
            inputPort, outputPort, 0, transitVlanId, OutputVlanType.NONE, meterId);

    assertEquals(
            scheme.ingressNoneFlowMod(inputPort, outputPort, transitVlanId, meterId, cookie),
            capture.getValue());
}

@Test
public void installEgressFlowNoneAction() throws Exception {
    Capture<OFFlowMod> capture = prepareForInstallTest();

    switchManager.installEgressFlow(dpid, cookieHex, cookie,
            inputPort, outputPort, transitVlanId, 0, OutputVlanType.NONE);

    assertEquals(
            scheme.egressNoneFlowMod(inputPort, outputPort, transitVlanId, cookie),
            capture.getValue());
}

@Test
public void installEgressFlowPushAction() throws Exception {
    Capture<OFFlowMod> capture = prepareForInstallTest();

    switchManager.installEgressFlow(dpid, cookieHex, cookie,
            inputPort, outputPort, transitVlanId, outputVlanId, OutputVlanType.PUSH);

    assertEquals(
            scheme.egressPushFlowMod(inputPort, outputPort, transitVlanId, outputVlanId, cookie),
            capture.getValue());
}

private Capture<List<OFMessage>> prepareChannelForWriteList() {
    EasyMock.expect(channel.isConnected()).andReturn(Boolean.TRUE).anyTimes();
    Capture<List<OFMessage>> cMsgList = new Capture<>();
    expect(channel.write(capture(cMsgList))).andReturn(null).once();
    replay(channel);
    return cMsgList;
}

@Test
public void testUserAgentPrefixAndSuffixAreAdded() throws Exception {
    String prefix = "somePrefix", suffix = "someSuffix";
    Request<?> request = mockRequest(SERVER_NAME, HttpMethodName.PUT, URI_NAME, true);

    HttpResponseHandler<AmazonWebServiceResponse<Object>> handler = createStubResponseHandler();
    EasyMock.replay(handler);
    ClientConfiguration config =
            new ClientConfiguration().withUserAgentPrefix(prefix).withUserAgentSuffix(suffix);

    Capture<HttpRequestBase> capturedRequest = new Capture<HttpRequestBase>();

    EasyMock.reset(httpClient);
    EasyMock
            .expect(httpClient.execute(
                    EasyMock.capture(capturedRequest), EasyMock.<HttpContext>anyObject()))
            .andReturn(createBasicHttpResponse())
            .once();
    EasyMock.replay(httpClient);

    AmazonHttpClient client = new AmazonHttpClient(config, httpClient, null);

    client.requestExecutionBuilder().request(request).execute(handler);

    String userAgent = capturedRequest.getValue().getFirstHeader("User-Agent").getValue();
    Assert.assertTrue(userAgent.startsWith(prefix));
    Assert.assertTrue(userAgent.endsWith(suffix));
}

@Test
public void installTransitFlow() throws Exception {
    Capture<OFFlowMod> capture = prepareForInstallTest();

    switchManager.installTransitFlow(dpid, cookieHex, cookie, inputPort, outputPort, transitVlanId);

    assertEquals(
            scheme.transitFlowMod(inputPort, outputPort, transitVlanId, cookie),
            capture.getValue());
}

@Test
public void installOneSwitchFlowReplaceAction() throws Exception {
    Capture<OFFlowMod> capture = prepareForInstallTest();

    switchManager.installOneSwitchFlow(dpid, cookieHex, cookie,
            inputPort, outputPort, inputVlanId, outputVlanId, OutputVlanType.REPLACE, meterId);

    assertEquals(
            scheme.oneSwitchReplaceFlowMod(inputPort, outputPort, inputVlanId, outputVlanId, meterId, cookie),
            capture.getValue());
}

@Test
public void installOneSwitchFlowPushAction() throws Exception {
    Capture<OFFlowMod> capture = prepareForInstallTest();

    switchManager.installOneSwitchFlow(dpid, cookieHex, cookie,
            inputPort, outputPort, 0, outputVlanId, OutputVlanType.PUSH, meterId);

    assertEquals(
            scheme.oneSwitchPushFlowMod(inputPort, outputPort, outputVlanId, meterId, cookie),
            capture.getValue());
}

@Test
public void installOneSwitchFlowPopAction() throws Exception {
    Capture<OFFlowMod> capture = prepareForInstallTest();

    switchManager.installOneSwitchFlow(dpid, cookieHex, cookie,
            inputPort, outputPort, inputVlanId, 0, OutputVlanType.POP, meterId);

    assertEquals(
            scheme.oneSwitchPopFlowMod(inputPort, outputPort, inputVlanId, meterId, cookie),
            capture.getValue());
}

@Test
public void installOneSwitchFlowNoneAction() throws Exception {
    Capture<OFFlowMod> capture = prepareForInstallTest();

    switchManager.installOneSwitchFlow(dpid, cookieHex, cookie,
            inputPort, outputPort, 0, 0, OutputVlanType.NONE, meterId);

    assertEquals(
            scheme.oneSwitchNoneFlowMod(inputPort, outputPort, meterId, cookie),
            capture.getValue());
}

/** write a request which triggers an OFErrorMsg response */
@Test(timeout = 5000)
public void testWriteRequestOFErrorMsg() throws InterruptedException, ExecutionException {
    Capture<List<OFMessage>> cMsgList = prepareChannelForWriteList();

    OFRoleRequest roleRequest = factory.buildRoleRequest().setRole(OFControllerRole.ROLE_MASTER).build();
    ListenableFuture<OFRoleReply> future = conn.writeRequest(roleRequest);
    assertThat("Connection should have 1 pending request",
            conn.getPendingRequestIds().size(), equalTo(1));

    assertThat("Should have captured MsgList", cMsgList.getValue(),
            Matchers.<OFMessage> contains(roleRequest));

    assertThat("Future should not be complete yet", future.isDone(), equalTo(false));
    OFRoleRequestFailedErrorMsg roleError = factory.errorMsgs().buildRoleRequestFailedErrorMsg()
        .setXid(roleRequest.getXid())
        .setCode(OFRoleRequestFailedCode.STALE)
        .build();

    assertThat("Connection should have accepted the response",
            conn.deliverResponse(roleError),
            equalTo(true));

    OFErrorMsgException e =
            FutureTestUtils.assertFutureFailedWithException(future,
                    OFErrorMsgException.class);
    assertThat(e.getErrorMessage(), CoreMatchers.<OFErrorMsg>equalTo(roleError));
}

@Test
public void testFloodBufferId() throws Exception {
    MockFloodlightProvider mockFloodlightProvider = getMockFloodlightProvider();
    this.packetIn = this.packetIn.createBuilder()
            .setBufferId(OFBufferId.of(10))
            .setXid(1)
            .build();

    OFActionOutput ao = OFFactories.getFactory(OFVersion.OF_13).actions().buildOutput().setPort(OFPort.FLOOD).build();
    List<OFAction> al = new ArrayList<OFAction>();
    al.add(ao);
    // build our expected flooded packetOut
    OFPacketOut po = OFFactories.getFactory(OFVersion.OF_13).buildPacketOut()
        .setActions(al)
        .setXid(1)
        .setBufferId(OFBufferId.of(10))
        .setInPort(OFPort.of(1))
        .build();

    // Mock up our expected behavior
    IOFSwitch mockSwitch = createMock(IOFSwitch.class);
    EasyMock.expect(mockSwitch.getOFFactory()).andReturn(OFFactories.getFactory(OFVersion.OF_13)).anyTimes();
    Capture<OFPacketOut> wc1 = new Capture<OFPacketOut>(CaptureType.ALL);
    mockSwitch.write(capture(wc1));

    // Start recording the replay on the mocks
    replay(mockSwitch);
    // Get the listener and trigger the packet in
    IOFMessageListener listener = mockFloodlightProvider.getListeners().get(
            OFType.PACKET_IN).get(0);
    listener.receive(mockSwitch, this.packetIn,
                     parseAndAnnotate(this.packetIn));

    // Verify the replay matched our expectations
    verify(mockSwitch);

    assertTrue(wc1.hasCaptured());
    OFMessage m = wc1.getValue();
    assertEquals(po, m);
}

private Capture<OFFlowMod> prepareForInstallTest() {
    Capture<OFFlowMod> capture = EasyMock.newCapture();

    expect(ofSwitchService.getSwitch(dpid)).andStubReturn(iofSwitch);
    expect(iofSwitch.getOFFactory()).andStubReturn(ofFactory);
    expect(iofSwitch.getSwitchDescription()).andStubReturn(switchDescription);
    expect(switchDescription.getManufacturerDescription()).andStubReturn("");
    expect(iofSwitch.write(capture(capture))).andReturn(true);
    EasyMock.expectLastCall();

    replay(ofSwitchService);
    replay(iofSwitch);
    replay(switchDescription);

    return capture;
}

@Test
public void testFlood() throws Exception {
    // build our expected flooded packetOut
    OFPacketOut po = factory.buildPacketOut()
        .setInPort(OFPort.of(1))
        .setActions(Arrays.asList((OFAction)factory.actions().output(OFPort.FLOOD, 0xffFFffFF)))
        .setBufferId(OFBufferId.NO_BUFFER)
        .setData(this.testPacketSerialized)
     .build();

    Capture<OFMessage> wc1 = new Capture<OFMessage>(CaptureType.ALL);

    // Mock up our expected behavior
    IOFSwitch mockSwitch = createMock(IOFSwitch.class);
    expect(mockSwitch.getId()).andReturn(DatapathId.of("00:11:22:33:44:55:66:77")).anyTimes();
    expect(mockSwitch.getOFFactory()).andReturn(factory).anyTimes();
    mockSwitch.write(EasyMock.capture(wc1)); // expect po
    EasyMock.expectLastCall().once();

    // Start recording the replay on the mocks
    replay(mockSwitch);
    // Get the listener and trigger the packet in
    IOFMessageListener listener = mockFloodlightProvider.getListeners().get(OFType.PACKET_IN).get(0);
    // Make sure it's the right listener
    listener.receive(mockSwitch, this.packetIn, parseAndAnnotate(this.packetIn));

    // Verify the replay matched our expectations
    OFPort result = learningSwitch.getFromPortMap(mockSwitch, MacAddress.of("00:44:33:22:11:00"), VlanVid.ofVlan(42));
    verify(mockSwitch);

    assertTrue(wc1.hasCaptured());
    assertTrue(OFMessageUtils.equalsIgnoreXid(wc1.getValue(), po));

    // Verify the MAC table inside the switch
    assertEquals(OFPort.of(1), result);
}

/**
 * Runs test case.
 *
 * @param value       data string from json resource file
 * @param flowCommand OFFlowAdd instance to compare result with
 * @throws InterruptedException if test was interrupted during run
 */
private void runTest(final String value, final OFFlowAdd flowCommand, final OFMeterMod meterCommand,
                     final OFFlowAdd reverseFlowCommand, final OFMeterMod reverseMeterCommand)
        throws InterruptedException {
    // construct kafka message
    ConsumerRecord<String, String> record = new ConsumerRecord<>("", 0, 0, "", value);

    // create parser instance
    KafkaMessageCollector.ParseRecord parseRecord = collector.new ParseRecord(record);
    // init test mocks
    Capture<OFFlowAdd> flowAddCapture = flowCommand == null ? null : newCapture(CaptureType.ALL);
    Capture<OFMeterMod> meterAddCapture = meterCommand == null ? null : newCapture(CaptureType.ALL);
    prepareMocks(flowAddCapture, meterAddCapture, reverseFlowCommand != null, reverseMeterCommand != null);

    // run parser and wait for termination or timeout
    parseRecordExecutor.execute(parseRecord);
    parseRecordExecutor.shutdown();
    parseRecordExecutor.awaitTermination(10, TimeUnit.SECONDS);

    // verify results
    if (meterCommand != null) {
        assertEquals(meterCommand, meterAddCapture.getValues().get(0));
        if (reverseMeterCommand != null) {
            assertEquals(reverseMeterCommand, meterAddCapture.getValues().get(1));
        }
    }
    if (flowCommand != null) {
        assertEquals(flowCommand, flowAddCapture.getValues().get(0));
        if (reverseFlowCommand != null) {
            assertEquals(reverseFlowCommand, flowAddCapture.getValues().get(1));
        }
    }
}

@Before
public void setUp() {
    broadcaster = createMock(TransactionBroadcaster.class);
    wallet = createMock(Wallet.class);
    connection = createMock(PaymentChannelServer.ServerConnection.class);
    serverVersionCapture = new Capture<TwoWayChannelMessage>();
    connection.sendToClient(capture(serverVersionCapture));
    Utils.setMockClock();
}