src/test/java/net/pterodactylus/fcp/quelaton/DefaultFcpClientTest.java

   1 package net.pterodactylus.fcp.quelaton;
   2
   3 import static org.hamcrest.MatcherAssert.assertThat;
   4 import static org.hamcrest.Matchers.allOf;
   5 import static org.hamcrest.Matchers.contains;
   6 import static org.hamcrest.Matchers.containsInAnyOrder;
   7 import static org.hamcrest.Matchers.hasItem;
   8 import static org.hamcrest.Matchers.hasSize;
   9 import static org.hamcrest.Matchers.is;
  10 import static org.hamcrest.Matchers.not;
  11 import static org.hamcrest.Matchers.notNullValue;
  12 import static org.hamcrest.Matchers.startsWith;
  13
  14 import java.io.ByteArrayInputStream;
  15 import java.io.File;
  16 import java.io.IOException;
  17 import java.net.URL;
  18 import java.nio.charset.StandardCharsets;
  19 import java.util.Arrays;
  20 import java.util.Collection;
  21 import java.util.List;
  22 import java.util.Optional;
  23 import java.util.concurrent.CopyOnWriteArrayList;
  24 import java.util.concurrent.CountDownLatch;
  25 import java.util.concurrent.ExecutionException;
  26 import java.util.concurrent.ExecutorService;
  27 import java.util.concurrent.Executors;
  28 import java.util.concurrent.Future;
  29 import java.util.concurrent.TimeUnit;
  30 import java.util.concurrent.atomic.AtomicBoolean;
  31 import java.util.concurrent.atomic.AtomicInteger;
  32 import java.util.function.Supplier;
  33 import java.util.stream.Collectors;
  34
  35 import net.pterodactylus.fcp.ARK;
  36 import net.pterodactylus.fcp.ConfigData;
  37 import net.pterodactylus.fcp.DSAGroup;
  38 import net.pterodactylus.fcp.FcpKeyPair;
  39 import net.pterodactylus.fcp.Key;
  40 import net.pterodactylus.fcp.NodeData;
  41 import net.pterodactylus.fcp.NodeRef;
  42 import net.pterodactylus.fcp.Peer;
  43 import net.pterodactylus.fcp.PeerNote;
  44 import net.pterodactylus.fcp.PluginInfo;
  45 import net.pterodactylus.fcp.Priority;
  46 import net.pterodactylus.fcp.fake.FakeTcpServer;
  47 import net.pterodactylus.fcp.quelaton.ClientGetCommand.Data;
  48
  49 import com.google.common.io.ByteStreams;
  50 import com.google.common.io.Files;
  51 import com.nitorcreations.junit.runners.NestedRunner;
  52 import org.hamcrest.Description;
  53 import org.hamcrest.Matcher;
  54 import org.hamcrest.Matchers;
  55 import org.hamcrest.TypeSafeDiagnosingMatcher;
  56 import org.junit.After;
  57 import org.junit.Assert;
  58 import org.junit.Test;
  59 import org.junit.runner.RunWith;
  60
  61 /**
  62  * Unit test for {@link DefaultFcpClient}.
  63  *
  64  * @author <a href="bombe@freenetproject.org">David ‘Bombe’ Roden</a>
  65  */
  66 @RunWith(NestedRunner.class)
  67 public class DefaultFcpClientTest {
  68
  69         private static final String INSERT_URI =
  70                 "SSK@RVCHbJdkkyTCeNN9AYukEg76eyqmiosSaNKgE3U9zUw,7SHH53gletBVb9JD7nBsyClbLQsBubDPEIcwg908r7Y,AQECAAE/";
  71         private static final String REQUEST_URI =
  72                 "SSK@wtbgd2loNcJCXvtQVOftl2tuWBomDQHfqS6ytpPRhfw,7SHH53gletBVb9JD7nBsyClbLQsBubDPEIcwg908r7Y,AQACAAE/";
  73
  74         private int threadCounter = 0;
  75         private final ExecutorService threadPool =
  76                 Executors.newCachedThreadPool(r -> new Thread(r, "Test-Thread-" + threadCounter++));
  77         private final FakeTcpServer fcpServer;
  78         private final DefaultFcpClient fcpClient;
  79
  80         public DefaultFcpClientTest() throws IOException {
  81                 fcpServer = new FakeTcpServer(threadPool);
  82                 fcpClient = new DefaultFcpClient(threadPool, "localhost", fcpServer.getPort(), () -> "Test");
  83         }
  84
  85         @After
  86         public void tearDown() throws IOException {
  87                 fcpServer.close();
  88                 threadPool.shutdown();
  89         }
  90
  91         private void connectNode() throws InterruptedException, ExecutionException, IOException {
  92                 fcpServer.connect().get();
  93                 fcpServer.collectUntil(is("EndMessage"));
  94                 fcpServer.writeLine("NodeHello",
  95                         "CompressionCodecs=4 - GZIP(0), BZIP2(1), LZMA(2), LZMA_NEW(3)",
  96                         "Revision=build01466",
  97                         "Testnet=false",
  98                         "Version=Fred,0.7,1.0,1466",
  99                         "Build=1466",
 100                         "ConnectionIdentifier=14318898267048452a81b36e7f13a3f0",
 101                         "Node=Fred",
 102                         "ExtBuild=29",
 103                         "FCPVersion=2.0",
 104                         "NodeLanguage=ENGLISH",
 105                         "ExtRevision=v29",
 106                         "EndMessage"
 107                 );
 108         }
 109
 110         private String extractIdentifier(List<String> lines) {
 111                 return lines.stream()
 112                         .filter(s -> s.startsWith("Identifier="))
 113                         .map(s -> s.substring(s.indexOf('=') + 1))
 114                         .findFirst()
 115                         .orElse("");
 116         }
 117
 118         private Matcher<List<String>> matchesFcpMessage(String name, String... requiredLines) {
 119                 return matchesFcpMessageWithTerminator(name, "EndMessage", requiredLines);
 120         }
 121
 122         private Matcher<Iterable<String>> hasHead(String firstElement) {
 123                 return new TypeSafeDiagnosingMatcher<Iterable<String>>() {
 124                         @Override
 125                         protected boolean matchesSafely(Iterable<String> iterable, Description mismatchDescription) {
 126                                 if (!iterable.iterator().hasNext()) {
 127                                         mismatchDescription.appendText("is empty");
 128                                         return false;
 129                                 }
 130                                 String element = iterable.iterator().next();
 131                                 if (!element.equals(firstElement)) {
 132                                         mismatchDescription.appendText("starts with ").appendValue(element);
 133                                         return false;
 134                                 }
 135                                 return true;
 136                         }
 137
 138                         @Override
 139                         public void describeTo(Description description) {
 140                                 description.appendText("starts with ").appendValue(firstElement);
 141                         }
 142                 };
 143         }
 144
 145         private Matcher<List<String>> matchesFcpMessageWithTerminator(
 146                 String name, String terminator, String... requiredLines) {
 147                 return allOf(hasHead(name), hasParameters(1, 1, requiredLines), hasTail(terminator));
 148         }
 149
 150         private Matcher<List<String>> hasParameters(int ignoreStart, int ignoreEnd, String... lines) {
 151                 return new TypeSafeDiagnosingMatcher<List<String>>() {
 152                         @Override
 153                         protected boolean matchesSafely(List<String> item, Description mismatchDescription) {
 154                                 if (item.size() < (ignoreStart + ignoreEnd)) {
 155                                         mismatchDescription.appendText("has only ").appendValue(item.size()).appendText(" elements");
 156                                         return false;
 157                                 }
 158                                 for (String line : lines) {
 159                                         if ((item.indexOf(line) < ignoreStart) || (item.indexOf(line) >= (item.size() - ignoreEnd))) {
 160                                                 mismatchDescription.appendText("does not contains ").appendValue(line);
 161                                                 return false;
 162                                         }
 163                                 }
 164                                 return true;
 165                         }
 166
 167                         @Override
 168                         public void describeTo(Description description) {
 169                                 description.appendText("contains ").appendValueList("(", ", ", ")", lines);
 170                                 description.appendText(", ignoring the first ").appendValue(ignoreStart);
 171                                 description.appendText(" and the last ").appendValue(ignoreEnd);
 172                         }
 173                 };
 174         }
 175
 176         private Matcher<List<String>> hasTail(String... lastElements) {
 177                 return new TypeSafeDiagnosingMatcher<List<String>>() {
 178                         @Override
 179                         protected boolean matchesSafely(List<String> list, Description mismatchDescription) {
 180                                 if (list.size() < lastElements.length) {
 181                                         mismatchDescription.appendText("is too small");
 182                                         return false;
 183                                 }
 184                                 List<String> tail = list.subList(list.size() - lastElements.length, list.size());
 185                                 if (!tail.equals(Arrays.asList(lastElements))) {
 186                                         mismatchDescription.appendText("ends with ").appendValueList("(", ", ", ")", tail);
 187                                         return false;
 188                                 }
 189                                 return true;
 190                         }
 191
 192                         @Override
 193                         public void describeTo(Description description) {
 194                                 description.appendText("ends with ").appendValueList("(", ", ", ")", lastElements);
 195                         }
 196                 };
 197         }
 198
 199         private List<String> lines;
 200         private String identifier;
 201
 202         private void connectAndAssert(Supplier<Matcher<List<String>>> requestMatcher)
 203         throws InterruptedException, ExecutionException, IOException {
 204                 connectNode();
 205                 readMessage(requestMatcher);
 206         }
 207
 208         private void readMessage(Supplier<Matcher<List<String>>> requestMatcher) throws IOException {
 209                 readMessage("EndMessage", requestMatcher);
 210         }
 211
 212         private void readMessage(String terminator, Supplier<Matcher<List<String>>> requestMatcher) throws IOException {
 213                 lines = fcpServer.collectUntil(is(terminator));
 214                 identifier = extractIdentifier(lines);
 215                 assertThat(lines, requestMatcher.get());
 216         }
 217
 218         private void replyWithProtocolError() throws IOException {
 219                 fcpServer.writeLine(
 220                         "ProtocolError",
 221                         "Identifier=" + identifier,
 222                         "EndMessage"
 223                 );
 224         }
 225
 226         public class ConnectionsAndKeyPairs {
 227
 228                 public class Connections {
 229
 230                         @Test(expected = ExecutionException.class)
 231                         public void throwsExceptionOnFailure() throws IOException, ExecutionException, InterruptedException {
 232                                 Future<FcpKeyPair> keyPairFuture = fcpClient.generateKeypair().execute();
 233                                 connectAndAssert(() -> matchesFcpMessage("GenerateSSK"));
 234                                 fcpServer.writeLine(
 235                                         "CloseConnectionDuplicateClientName",
 236                                         "EndMessage"
 237                                 );
 238                                 keyPairFuture.get();
 239                         }
 240
 241                         @Test(expected = ExecutionException.class)
 242                         public void throwsExceptionIfConnectionIsClosed() throws IOException, ExecutionException, InterruptedException {
 243                                 Future<FcpKeyPair> keyPairFuture = fcpClient.generateKeypair().execute();
 244                                 connectAndAssert(() -> matchesFcpMessage("GenerateSSK"));
 245                                 fcpServer.close();
 246                                 keyPairFuture.get();
 247                         }
 248
 249                         @Test
 250                         public void connectionIsReused() throws InterruptedException, ExecutionException, IOException {
 251                                 Future<FcpKeyPair> keyPair = fcpClient.generateKeypair().execute();
 252                                 connectAndAssert(() -> matchesFcpMessage("GenerateSSK"));
 253                                 replyWithKeyPair();
 254                                 keyPair.get();
 255                                 keyPair = fcpClient.generateKeypair().execute();
 256                                 readMessage(() -> matchesFcpMessage("GenerateSSK"));
 257                                 identifier = extractIdentifier(lines);
 258                                 replyWithKeyPair();
 259                                 keyPair.get();
 260                         }
 261
 262                         @Test
 263                         public void defaultFcpClientCanReconnectAfterConnectionHasBeenClosed()
 264                         throws InterruptedException, ExecutionException, IOException {
 265                                 Future<FcpKeyPair> keyPair = fcpClient.generateKeypair().execute();
 266                                 connectAndAssert(() -> matchesFcpMessage("GenerateSSK"));
 267                                 fcpServer.close();
 268                                 try {
 269                                         keyPair.get();
 270                                         Assert.fail();
 271                                 } catch (ExecutionException e) {
 272                                         /* ignore. */
 273                                 }
 274                                 keyPair = fcpClient.generateKeypair().execute();
 275                                 connectAndAssert(() -> matchesFcpMessage("GenerateSSK"));
 276                                 replyWithKeyPair();
 277                                 keyPair.get();
 278                         }
 279
 280                 }
 281
 282                 public class GenerateKeyPair {
 283
 284                         @Test
 285                         public void defaultFcpClientCanGenerateKeypair()
 286                         throws ExecutionException, InterruptedException, IOException {
 287                                 Future<FcpKeyPair> keyPairFuture = fcpClient.generateKeypair().execute();
 288                                 connectAndAssert(() -> matchesFcpMessage("GenerateSSK"));
 289                                 replyWithKeyPair();
 290                                 FcpKeyPair keyPair = keyPairFuture.get();
 291                                 assertThat(keyPair.getPublicKey(), is(REQUEST_URI));
 292                                 assertThat(keyPair.getPrivateKey(), is(INSERT_URI));
 293                         }
 294
 295                 }
 296
 297                 private void replyWithKeyPair() throws IOException {
 298                         fcpServer.writeLine("SSKKeypair",
 299                                 "InsertURI=" + INSERT_URI + "",
 300                                 "RequestURI=" + REQUEST_URI + "",
 301                                 "Identifier=" + identifier,
 302                                 "EndMessage");
 303                 }
 304
 305         }
 306
 307         public class Peers {
 308
 309                 public class PeerCommands {
 310
 311                         public class ListPeer {
 312
 313                                 @Test
 314                                 public void byIdentity() throws InterruptedException, ExecutionException, IOException {
 315                                         Future<Optional<Peer>> peer = fcpClient.listPeer().byIdentity("id1").execute();
 316                                         connectAndAssert(() -> matchesListPeer("id1"));
 317                                         replyWithPeer("id1");
 318                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 319                                 }
 320
 321                                 @Test
 322                                 public void byHostAndPort() throws InterruptedException, ExecutionException, IOException {
 323                                         Future<Optional<Peer>> peer = fcpClient.listPeer().byHostAndPort("host.free.net", 12345).execute();
 324                                         connectAndAssert(() -> matchesListPeer("host.free.net:12345"));
 325                                         replyWithPeer("id1");
 326                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 327                                 }
 328
 329                                 @Test
 330                                 public void byName() throws InterruptedException, ExecutionException, IOException {
 331                                         Future<Optional<Peer>> peer = fcpClient.listPeer().byName("FriendNode").execute();
 332                                         connectAndAssert(() -> matchesListPeer("FriendNode"));
 333                                         replyWithPeer("id1");
 334                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 335                                 }
 336
 337                                 @Test
 338                                 public void unknownNodeIdentifier() throws InterruptedException, ExecutionException, IOException {
 339                                         Future<Optional<Peer>> peer = fcpClient.listPeer().byIdentity("id2").execute();
 340                                         connectAndAssert(() -> matchesListPeer("id2"));
 341                                         replyWithUnknownNodeIdentifier();
 342                                         assertThat(peer.get().isPresent(), is(false));
 343                                 }
 344
 345                                 private Matcher<List<String>> matchesListPeer(String nodeId) {
 346                                         return matchesFcpMessage(
 347                                                 "ListPeer",
 348                                                 "Identifier=" + identifier,
 349                                                 "NodeIdentifier=" + nodeId
 350                                         );
 351                                 }
 352
 353                         }
 354
 355                         public class ListPeers {
 356
 357                                 @Test
 358                                 public void withoutMetadataOrVolatile() throws IOException, ExecutionException, InterruptedException {
 359                                         Future<Collection<Peer>> peers = fcpClient.listPeers().execute();
 360                                         connectAndAssert(() -> matchesListPeers(false, false));
 361                                         replyWithPeer("id1");
 362                                         replyWithPeer("id2");
 363                                         sendEndOfPeerList();
 364                                         assertThat(peers.get(), hasSize(2));
 365                                         assertThat(peers.get().stream().map(Peer::getIdentity).collect(Collectors.toList()),
 366                                                 containsInAnyOrder("id1", "id2"));
 367                                 }
 368
 369                                 @Test
 370                                 public void withMetadata() throws IOException, ExecutionException, InterruptedException {
 371                                         Future<Collection<Peer>> peers = fcpClient.listPeers().includeMetadata().execute();
 372                                         connectAndAssert(() -> matchesListPeers(false, true));
 373                                         replyWithPeer("id1", "metadata.foo=bar1");
 374                                         replyWithPeer("id2", "metadata.foo=bar2");
 375                                         sendEndOfPeerList();
 376                                         assertThat(peers.get(), hasSize(2));
 377                                         assertThat(peers.get().stream().map(peer -> peer.getMetadata("foo")).collect(Collectors.toList()),
 378                                                 containsInAnyOrder("bar1", "bar2"));
 379                                 }
 380
 381                                 @Test
 382                                 public void withVolatile() throws IOException, ExecutionException, InterruptedException {
 383                                         Future<Collection<Peer>> peers = fcpClient.listPeers().includeVolatile().execute();
 384                                         connectAndAssert(() -> matchesListPeers(true, false));
 385                                         replyWithPeer("id1", "volatile.foo=bar1");
 386                                         replyWithPeer("id2", "volatile.foo=bar2");
 387                                         sendEndOfPeerList();
 388                                         assertThat(peers.get(), hasSize(2));
 389                                         assertThat(peers.get().stream().map(peer -> peer.getVolatile("foo")).collect(Collectors.toList()),
 390                                                 containsInAnyOrder("bar1", "bar2"));
 391                                 }
 392
 393                                 private Matcher<List<String>> matchesListPeers(boolean withVolatile, boolean withMetadata) {
 394                                         return matchesFcpMessage(
 395                                                 "ListPeers",
 396                                                 "WithVolatile=" + withVolatile,
 397                                                 "WithMetadata=" + withMetadata
 398                                         );
 399                                 }
 400
 401                                 private void sendEndOfPeerList() throws IOException {
 402                                         fcpServer.writeLine(
 403                                                 "EndListPeers",
 404                                                 "Identifier=" + identifier,
 405                                                 "EndMessage"
 406                                         );
 407                                 }
 408
 409                         }
 410
 411                         public class AddPeer {
 412
 413                                 @Test
 414                                 public void fromFile() throws InterruptedException, ExecutionException, IOException {
 415                                         Future<Optional<Peer>> peer = fcpClient.addPeer().fromFile(new File("/tmp/ref.txt")).execute();
 416                                         connectAndAssert(() -> allOf(matchesAddPeer(), hasItem("File=/tmp/ref.txt")));
 417                                         replyWithPeer("id1");
 418                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 419                                 }
 420
 421                                 @Test
 422                                 public void fromUrl() throws InterruptedException, ExecutionException, IOException {
 423                                         Future<Optional<Peer>> peer = fcpClient.addPeer().fromURL(new URL("http://node.ref/")).execute();
 424                                         connectAndAssert(() -> allOf(matchesAddPeer(), hasItem("URL=http://node.ref/")));
 425                                         replyWithPeer("id1");
 426                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 427                                 }
 428
 429                                 @Test
 430                                 public void fromNodeRef() throws InterruptedException, ExecutionException, IOException {
 431                                         NodeRef nodeRef = createNodeRef();
 432                                         Future<Optional<Peer>> peer = fcpClient.addPeer().fromNodeRef(nodeRef).execute();
 433                                         connectAndAssert(() -> allOf(matchesAddPeer(), Matchers.<String>hasItems(
 434                                                 "myName=name",
 435                                                 "ark.pubURI=public",
 436                                                 "ark.number=1",
 437                                                 "dsaGroup.g=base",
 438                                                 "dsaGroup.p=prime",
 439                                                 "dsaGroup.q=subprime",
 440                                                 "dsaPubKey.y=dsa-public",
 441                                                 "physical.udp=1.2.3.4:5678",
 442                                                 "auth.negTypes=3;5",
 443                                                 "sig=sig"
 444                                         )));
 445                                         replyWithPeer("id1");
 446                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 447                                 }
 448
 449                                 @Test
 450                                 public void protocolErrorEndsCommand() throws InterruptedException, ExecutionException, IOException {
 451                                         Future<Optional<Peer>> peer = fcpClient.addPeer().fromFile(new File("/tmp/ref.txt")).execute();
 452                                         connectAndAssert(() -> allOf(matchesAddPeer(), hasItem("File=/tmp/ref.txt")));
 453                                         replyWithProtocolError();
 454                                         assertThat(peer.get().isPresent(), is(false));
 455                                 }
 456
 457                                 private NodeRef createNodeRef() {
 458                                         NodeRef nodeRef = new NodeRef();
 459                                         nodeRef.setIdentity("id1");
 460                                         nodeRef.setName("name");
 461                                         nodeRef.setARK(new ARK("public", "1"));
 462                                         nodeRef.setDSAGroup(new DSAGroup("base", "prime", "subprime"));
 463                                         nodeRef.setNegotiationTypes(new int[] { 3, 5 });
 464                                         nodeRef.setPhysicalUDP("1.2.3.4:5678");
 465                                         nodeRef.setDSAPublicKey("dsa-public");
 466                                         nodeRef.setSignature("sig");
 467                                         return nodeRef;
 468                                 }
 469
 470                                 private Matcher<List<String>> matchesAddPeer() {
 471                                         return matchesFcpMessage(
 472                                                 "AddPeer",
 473                                                 "Identifier=" + identifier
 474                                         );
 475                                 }
 476
 477                         }
 478
 479                         public class ModifyPeer {
 480
 481                                 @Test
 482                                 public void defaultFcpClientCanEnablePeerByName()
 483                                 throws InterruptedException, ExecutionException, IOException {
 484                                         Future<Optional<Peer>> peer = fcpClient.modifyPeer().enable().byName("id1").execute();
 485                                         connectAndAssert(() -> matchesModifyPeer("id1", "IsDisabled", false));
 486                                         replyWithPeer("id1");
 487                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 488                                 }
 489
 490                                 @Test
 491                                 public void defaultFcpClientCanDisablePeerByName()
 492                                 throws InterruptedException, ExecutionException, IOException {
 493                                         Future<Optional<Peer>> peer = fcpClient.modifyPeer().disable().byName("id1").execute();
 494                                         connectAndAssert(() -> matchesModifyPeer("id1", "IsDisabled", true));
 495                                         replyWithPeer("id1");
 496                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 497                                 }
 498
 499                                 @Test
 500                                 public void defaultFcpClientCanEnablePeerByIdentity()
 501                                 throws InterruptedException, ExecutionException, IOException {
 502                                         Future<Optional<Peer>> peer = fcpClient.modifyPeer().enable().byIdentity("id1").execute();
 503                                         connectAndAssert(() -> matchesModifyPeer("id1", "IsDisabled", false));
 504                                         replyWithPeer("id1");
 505                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 506                                 }
 507
 508                                 @Test
 509                                 public void defaultFcpClientCanEnablePeerByHostAndPort()
 510                                 throws InterruptedException, ExecutionException, IOException {
 511                                         Future<Optional<Peer>> peer =
 512                                                 fcpClient.modifyPeer().enable().byHostAndPort("1.2.3.4", 5678).execute();
 513                                         connectAndAssert(() -> matchesModifyPeer("1.2.3.4:5678", "IsDisabled", false));
 514                                         replyWithPeer("id1");
 515                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 516                                 }
 517
 518                                 @Test
 519                                 public void allowLocalAddressesOfPeer() throws InterruptedException, ExecutionException, IOException {
 520                                         Future<Optional<Peer>> peer =
 521                                                 fcpClient.modifyPeer().allowLocalAddresses().byIdentity("id1").execute();
 522                                         connectAndAssert(() -> allOf(
 523                                                 matchesModifyPeer("id1", "AllowLocalAddresses", true),
 524                                                 not(contains(startsWith("IsDisabled=")))
 525                                         ));
 526                                         replyWithPeer("id1");
 527                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 528                                 }
 529
 530                                 @Test
 531                                 public void disallowLocalAddressesOfPeer()
 532                                 throws InterruptedException, ExecutionException, IOException {
 533                                         Future<Optional<Peer>> peer =
 534                                                 fcpClient.modifyPeer().disallowLocalAddresses().byIdentity("id1").execute();
 535                                         connectAndAssert(() -> allOf(
 536                                                 matchesModifyPeer("id1", "AllowLocalAddresses", false),
 537                                                 not(contains(startsWith("IsDisabled=")))
 538                                         ));
 539                                         replyWithPeer("id1");
 540                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 541                                 }
 542
 543                                 @Test
 544                                 public void setBurstOnlyForPeer() throws InterruptedException, ExecutionException, IOException {
 545                                         Future<Optional<Peer>> peer = fcpClient.modifyPeer().setBurstOnly().byIdentity("id1").execute();
 546                                         connectAndAssert(() -> allOf(
 547                                                 matchesModifyPeer("id1", "IsBurstOnly", true),
 548                                                 not(contains(startsWith("AllowLocalAddresses="))),
 549                                                 not(contains(startsWith("IsDisabled=")))
 550                                         ));
 551                                         replyWithPeer("id1");
 552                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 553                                 }
 554
 555                                 @Test
 556                                 public void clearBurstOnlyForPeer() throws InterruptedException, ExecutionException, IOException {
 557                                         Future<Optional<Peer>> peer = fcpClient.modifyPeer().clearBurstOnly().byIdentity("id1").execute();
 558                                         connectAndAssert(() -> allOf(
 559                                                 matchesModifyPeer("id1", "IsBurstOnly", false),
 560                                                 not(contains(startsWith("AllowLocalAddresses="))),
 561                                                 not(contains(startsWith("IsDisabled=")))
 562                                         ));
 563                                         replyWithPeer("id1");
 564                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 565                                 }
 566
 567                                 @Test
 568                                 public void defaultFcpClientCanSetListenOnlyForPeer()
 569                                 throws InterruptedException, ExecutionException, IOException {
 570                                         Future<Optional<Peer>> peer = fcpClient.modifyPeer().setListenOnly().byIdentity("id1").execute();
 571                                         connectAndAssert(() -> allOf(
 572                                                 matchesModifyPeer("id1", "IsListenOnly", true),
 573                                                 not(contains(startsWith("AllowLocalAddresses="))),
 574                                                 not(contains(startsWith("IsDisabled="))),
 575                                                 not(contains(startsWith("IsBurstOnly=")))
 576                                         ));
 577                                         replyWithPeer("id1");
 578                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 579                                 }
 580
 581                                 @Test
 582                                 public void clearListenOnlyForPeer() throws InterruptedException, ExecutionException, IOException {
 583                                         Future<Optional<Peer>> peer = fcpClient.modifyPeer().clearListenOnly().byIdentity("id1").execute();
 584                                         connectAndAssert(() -> allOf(
 585                                                 matchesModifyPeer("id1", "IsListenOnly", false),
 586                                                 not(contains(startsWith("AllowLocalAddresses="))),
 587                                                 not(contains(startsWith("IsDisabled="))),
 588                                                 not(contains(startsWith("IsBurstOnly=")))
 589                                         ));
 590                                         replyWithPeer("id1");
 591                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 592                                 }
 593
 594                                 @Test
 595                                 public void ignoreSourceForPeer() throws InterruptedException, ExecutionException, IOException {
 596                                         Future<Optional<Peer>> peer = fcpClient.modifyPeer().ignoreSource().byIdentity("id1").execute();
 597                                         connectAndAssert(() -> allOf(
 598                                                 matchesModifyPeer("id1", "IgnoreSourcePort", true),
 599                                                 not(contains(startsWith("AllowLocalAddresses="))),
 600                                                 not(contains(startsWith("IsDisabled="))),
 601                                                 not(contains(startsWith("IsBurstOnly="))),
 602                                                 not(contains(startsWith("IsListenOnly=")))
 603                                         ));
 604                                         replyWithPeer("id1");
 605                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 606                                 }
 607
 608                                 @Test
 609                                 public void useSourceForPeer() throws InterruptedException, ExecutionException, IOException {
 610                                         Future<Optional<Peer>> peer = fcpClient.modifyPeer().useSource().byIdentity("id1").execute();
 611                                         connectAndAssert(() -> allOf(
 612                                                 matchesModifyPeer("id1", "IgnoreSourcePort", false),
 613                                                 not(contains(startsWith("AllowLocalAddresses="))),
 614                                                 not(contains(startsWith("IsDisabled="))),
 615                                                 not(contains(startsWith("IsBurstOnly="))),
 616                                                 not(contains(startsWith("IsListenOnly=")))
 617                                         ));
 618                                         replyWithPeer("id1");
 619                                         assertThat(peer.get().get().getIdentity(), is("id1"));
 620                                 }
 621
 622                                 @Test
 623                                 public void unknownNode() throws InterruptedException, ExecutionException, IOException {
 624                                         Future<Optional<Peer>> peer = fcpClient.modifyPeer().enable().byIdentity("id1").execute();
 625                                         connectAndAssert(() -> matchesModifyPeer("id1", "IsDisabled", false));
 626                                         replyWithUnknownNodeIdentifier();
 627                                         assertThat(peer.get().isPresent(), is(false));
 628                                 }
 629
 630                                 private Matcher<List<String>> matchesModifyPeer(String nodeIdentifier, String setting, boolean value) {
 631                                         return matchesFcpMessage(
 632                                                 "ModifyPeer",
 633                                                 "Identifier=" + identifier,
 634                                                 "NodeIdentifier=" + nodeIdentifier,
 635                                                 setting + "=" + value
 636                                         );
 637                                 }
 638
 639                         }
 640
 641                         public class RemovePeer {
 642
 643                                 @Test
 644                                 public void byName() throws InterruptedException, ExecutionException, IOException {
 645                                         Future<Boolean> peer = fcpClient.removePeer().byName("Friend1").execute();
 646                                         connectAndAssert(() -> matchesRemovePeer("Friend1"));
 647                                         replyWithPeerRemoved("Friend1");
 648                                         assertThat(peer.get(), is(true));
 649                                 }
 650
 651                                 @Test
 652                                 public void invalidName() throws InterruptedException, ExecutionException, IOException {
 653                                         Future<Boolean> peer = fcpClient.removePeer().byName("NotFriend1").execute();
 654                                         connectAndAssert(() -> matchesRemovePeer("NotFriend1"));
 655                                         replyWithUnknownNodeIdentifier();
 656                                         assertThat(peer.get(), is(false));
 657                                 }
 658
 659                                 @Test
 660                                 public void byIdentity() throws InterruptedException, ExecutionException, IOException {
 661                                         Future<Boolean> peer = fcpClient.removePeer().byIdentity("id1").execute();
 662                                         connectAndAssert(() -> matchesRemovePeer("id1"));
 663                                         replyWithPeerRemoved("id1");
 664                                         assertThat(peer.get(), is(true));
 665                                 }
 666
 667                                 @Test
 668                                 public void byHostAndPort() throws InterruptedException, ExecutionException, IOException {
 669                                         Future<Boolean> peer = fcpClient.removePeer().byHostAndPort("1.2.3.4", 5678).execute();
 670                                         connectAndAssert(() -> matchesRemovePeer("1.2.3.4:5678"));
 671                                         replyWithPeerRemoved("Friend1");
 672                                         assertThat(peer.get(), is(true));
 673                                 }
 674
 675                                 private Matcher<List<String>> matchesRemovePeer(String nodeIdentifier) {
 676                                         return matchesFcpMessage(
 677                                                 "RemovePeer",
 678                                                 "Identifier=" + identifier,
 679                                                 "NodeIdentifier=" + nodeIdentifier
 680                                         );
 681                                 }
 682
 683                                 private void replyWithPeerRemoved(String nodeIdentifier) throws IOException {
 684                                         fcpServer.writeLine(
 685                                                 "PeerRemoved",
 686                                                 "Identifier=" + identifier,
 687                                                 "NodeIdentifier=" + nodeIdentifier,
 688                                                 "EndMessage"
 689                                         );
 690                                 }
 691
 692                         }
 693
 694                         private void replyWithPeer(String peerId, String... additionalLines) throws IOException {
 695                                 fcpServer.writeLine(
 696                                         "Peer",
 697                                         "Identifier=" + identifier,
 698                                         "identity=" + peerId,
 699                                         "opennet=false",
 700                                         "ark.pubURI=SSK@3YEf.../ark",
 701                                         "ark.number=78",
 702                                         "auth.negTypes=2",
 703                                         "version=Fred,0.7,1.0,1466",
 704                                         "lastGoodVersion=Fred,0.7,1.0,1466"
 705                                 );
 706                                 fcpServer.writeLine(additionalLines);
 707                                 fcpServer.writeLine("EndMessage");
 708                         }
 709
 710                 }
 711
 712                 public class PeerNoteCommands {
 713
 714                         public class ListPeerNotes {
 715
 716                                 @Test
 717                                 public void onUnknownNodeIdentifier() throws InterruptedException, ExecutionException, IOException {
 718                                         Future<Optional<PeerNote>> peerNote = fcpClient.listPeerNotes().byName("Friend1").execute();
 719                                         connectAndAssert(() -> matchesListPeerNotes("Friend1"));
 720                                         replyWithUnknownNodeIdentifier();
 721                                         assertThat(peerNote.get().isPresent(), is(false));
 722                                 }
 723
 724                                 @Test
 725                                 public void byNodeName() throws InterruptedException, ExecutionException, IOException {
 726                                         Future<Optional<PeerNote>> peerNote = fcpClient.listPeerNotes().byName("Friend1").execute();
 727                                         connectAndAssert(() -> matchesListPeerNotes("Friend1"));
 728                                         replyWithPeerNote();
 729                                         replyWithEndListPeerNotes();
 730                                         assertThat(peerNote.get().get().getNoteText(), is("RXhhbXBsZSBUZXh0Lg=="));
 731                                         assertThat(peerNote.get().get().getPeerNoteType(), is(1));
 732                                 }
 733
 734                                 @Test
 735                                 public void byNodeIdentifier() throws InterruptedException, ExecutionException, IOException {
 736                                         Future<Optional<PeerNote>> peerNote = fcpClient.listPeerNotes().byIdentity("id1").execute();
 737                                         connectAndAssert(() -> matchesListPeerNotes("id1"));
 738                                         replyWithPeerNote();
 739                                         replyWithEndListPeerNotes();
 740                                         assertThat(peerNote.get().get().getNoteText(), is("RXhhbXBsZSBUZXh0Lg=="));
 741                                         assertThat(peerNote.get().get().getPeerNoteType(), is(1));
 742                                 }
 743
 744                                 @Test
 745                                 public void byHostAndPort() throws InterruptedException, ExecutionException, IOException {
 746                                         Future<Optional<PeerNote>> peerNote =
 747                                                 fcpClient.listPeerNotes().byHostAndPort("1.2.3.4", 5678).execute();
 748                                         connectAndAssert(() -> matchesListPeerNotes("1.2.3.4:5678"));
 749                                         replyWithPeerNote();
 750                                         replyWithEndListPeerNotes();
 751                                         assertThat(peerNote.get().get().getNoteText(), is("RXhhbXBsZSBUZXh0Lg=="));
 752                                         assertThat(peerNote.get().get().getPeerNoteType(), is(1));
 753                                 }
 754
 755                                 private Matcher<List<String>> matchesListPeerNotes(String nodeIdentifier) {
 756                                         return matchesFcpMessage(
 757                                                 "ListPeerNotes",
 758                                                 "NodeIdentifier=" + nodeIdentifier
 759                                         );
 760                                 }
 761
 762                                 private void replyWithEndListPeerNotes() throws IOException {
 763                                         fcpServer.writeLine(
 764                                                 "EndListPeerNotes",
 765                                                 "Identifier=" + identifier,
 766                                                 "EndMessage"
 767                                         );
 768                                 }
 769
 770                                 private void replyWithPeerNote() throws IOException {
 771                                         fcpServer.writeLine(
 772                                                 "PeerNote",
 773                                                 "Identifier=" + identifier,
 774                                                 "NodeIdentifier=Friend1",
 775                                                 "NoteText=RXhhbXBsZSBUZXh0Lg==",
 776                                                 "PeerNoteType=1",
 777                                                 "EndMessage"
 778                                         );
 779                                 }
 780
 781                         }
 782
 783                         public class ModifyPeerNotes {
 784
 785                                 @Test
 786                                 public void byName() throws InterruptedException, ExecutionException, IOException {
 787                                         Future<Boolean> noteUpdated =
 788                                                 fcpClient.modifyPeerNote().darknetComment("foo").byName("Friend1").execute();
 789                                         connectAndAssert(() -> matchesModifyPeerNote("Friend1"));
 790                                         replyWithPeerNote();
 791                                         assertThat(noteUpdated.get(), is(true));
 792                                 }
 793
 794                                 @Test
 795                                 public void onUnknownNodeIdentifier() throws InterruptedException, ExecutionException, IOException {
 796                                         Future<Boolean> noteUpdated =
 797                                                 fcpClient.modifyPeerNote().darknetComment("foo").byName("Friend1").execute();
 798                                         connectAndAssert(() -> matchesModifyPeerNote("Friend1"));
 799                                         replyWithUnknownNodeIdentifier();
 800                                         assertThat(noteUpdated.get(), is(false));
 801                                 }
 802
 803                                 @Test
 804                                 public void defaultFcpClientFailsToModifyPeerNoteWithoutPeerNote()
 805                                 throws InterruptedException, ExecutionException, IOException {
 806                                         Future<Boolean> noteUpdated = fcpClient.modifyPeerNote().byName("Friend1").execute();
 807                                         assertThat(noteUpdated.get(), is(false));
 808                                 }
 809
 810                                 @Test
 811                                 public void byIdentifier() throws InterruptedException, ExecutionException, IOException {
 812                                         Future<Boolean> noteUpdated =
 813                                                 fcpClient.modifyPeerNote().darknetComment("foo").byIdentifier("id1").execute();
 814                                         connectAndAssert(() -> matchesModifyPeerNote("id1"));
 815                                         replyWithPeerNote();
 816                                         assertThat(noteUpdated.get(), is(true));
 817                                 }
 818
 819                                 @Test
 820                                 public void byHostAndPort() throws InterruptedException, ExecutionException, IOException {
 821                                         Future<Boolean> noteUpdated =
 822                                                 fcpClient.modifyPeerNote().darknetComment("foo").byHostAndPort("1.2.3.4", 5678).execute();
 823                                         connectAndAssert(() -> matchesModifyPeerNote("1.2.3.4:5678"));
 824                                         replyWithPeerNote();
 825                                         assertThat(noteUpdated.get(), is(true));
 826                                 }
 827
 828                                 private Matcher<List<String>> matchesModifyPeerNote(String nodeIdentifier) {
 829                                         return matchesFcpMessage(
 830                                                 "ModifyPeerNote",
 831                                                 "Identifier=" + identifier,
 832                                                 "NodeIdentifier=" + nodeIdentifier,
 833                                                 "PeerNoteType=1",
 834                                                 "NoteText=Zm9v"
 835                                         );
 836                                 }
 837
 838                                 private void replyWithPeerNote() throws IOException {
 839                                         fcpServer.writeLine(
 840                                                 "PeerNote",
 841                                                 "Identifier=" + identifier,
 842                                                 "NodeIdentifier=Friend1",
 843                                                 "NoteText=Zm9v",
 844                                                 "PeerNoteType=1",
 845                                                 "EndMessage"
 846                                         );
 847                                 }
 848
 849                         }
 850
 851                 }
 852
 853                 private void replyWithUnknownNodeIdentifier() throws IOException {
 854                         fcpServer.writeLine(
 855                                 "UnknownNodeIdentifier",
 856                                 "Identifier=" + identifier,
 857                                 "NodeIdentifier=id2",
 858                                 "EndMessage"
 859                         );
 860                 }
 861
 862         }
 863
 864         public class PluginCommands {
 865
 866                 private static final String CLASS_NAME = "foo.plugin.Plugin";
 867
 868                 private void replyWithPluginInfo() throws IOException {
 869                         fcpServer.writeLine(
 870                                 "PluginInfo",
 871                                 "Identifier=" + identifier,
 872                                 "PluginName=superPlugin",
 873                                 "IsTalkable=true",
 874                                 "LongVersion=1.2.3",
 875                                 "Version=42",
 876                                 "OriginUri=superPlugin",
 877                                 "Started=true",
 878                                 "EndMessage"
 879                         );
 880                 }
 881
 882                 private void verifyPluginInfo(Future<Optional<PluginInfo>> pluginInfo)
 883                 throws InterruptedException, ExecutionException {
 884                         assertThat(pluginInfo.get().get().getPluginName(), is("superPlugin"));
 885                         assertThat(pluginInfo.get().get().getOriginalURI(), is("superPlugin"));
 886                         assertThat(pluginInfo.get().get().isTalkable(), is(true));
 887                         assertThat(pluginInfo.get().get().getVersion(), is("42"));
 888                         assertThat(pluginInfo.get().get().getLongVersion(), is("1.2.3"));
 889                         assertThat(pluginInfo.get().get().isStarted(), is(true));
 890                 }
 891
 892                 public class LoadPlugin {
 893
 894                         public class OfficialPlugins {
 895
 896                                 @Test
 897                                 public void fromFreenet() throws ExecutionException, InterruptedException, IOException {
 898                                         Future<Optional<PluginInfo>> pluginInfo =
 899                                                 fcpClient.loadPlugin().officialFromFreenet("superPlugin").execute();
 900                                         connectAndAssert(() -> createMatcherForOfficialSource("freenet"));
 901                                         assertThat(lines, not(contains(startsWith("Store="))));
 902                                         replyWithPluginInfo();
 903                                         verifyPluginInfo(pluginInfo);
 904                                 }
 905
 906                                 @Test
 907                                 public void persistentFromFreenet() throws ExecutionException, InterruptedException, IOException {
 908                                         Future<Optional<PluginInfo>> pluginInfo =
 909                                                 fcpClient.loadPlugin().addToConfig().officialFromFreenet("superPlugin").execute();
 910                                         connectAndAssert(() -> createMatcherForOfficialSource("freenet"));
 911                                         assertThat(lines, hasItem("Store=true"));
 912                                         replyWithPluginInfo();
 913                                         verifyPluginInfo(pluginInfo);
 914                                 }
 915
 916                                 @Test
 917                                 public void fromHttps() throws ExecutionException, InterruptedException, IOException {
 918                                         Future<Optional<PluginInfo>> pluginInfo =
 919                                                 fcpClient.loadPlugin().officialFromHttps("superPlugin").execute();
 920                                         connectAndAssert(() -> createMatcherForOfficialSource("https"));
 921                                         replyWithPluginInfo();
 922                                         verifyPluginInfo(pluginInfo);
 923                                 }
 924
 925                                 private Matcher<List<String>> createMatcherForOfficialSource(String officialSource) {
 926                                         return matchesFcpMessage(
 927                                                 "LoadPlugin",
 928                                                 "Identifier=" + identifier,
 929                                                 "PluginURL=superPlugin",
 930                                                 "URLType=official",
 931                                                 "OfficialSource=" + officialSource
 932                                         );
 933                                 }
 934
 935                         }
 936
 937                         public class FromOtherSources {
 938
 939                                 private static final String FILE_PATH = "/path/to/plugin.jar";
 940                                 private static final String URL = "http://server.com/plugin.jar";
 941                                 private static final String KEY = "KSK@plugin.jar";
 942
 943                                 @Test
 944                                 public void fromFile() throws ExecutionException, InterruptedException, IOException {
 945                                         Future<Optional<PluginInfo>> pluginInfo = fcpClient.loadPlugin().fromFile(FILE_PATH).execute();
 946                                         connectAndAssert(() -> createMatcher("file", FILE_PATH));
 947                                         replyWithPluginInfo();
 948                                         verifyPluginInfo(pluginInfo);
 949                                 }
 950
 951                                 @Test
 952                                 public void fromUrl() throws ExecutionException, InterruptedException, IOException {
 953                                         Future<Optional<PluginInfo>> pluginInfo = fcpClient.loadPlugin().fromUrl(URL).execute();
 954                                         connectAndAssert(() -> createMatcher("url", URL));
 955                                         replyWithPluginInfo();
 956                                         verifyPluginInfo(pluginInfo);
 957                                 }
 958
 959                                 @Test
 960                                 public void fromFreenet() throws ExecutionException, InterruptedException, IOException {
 961                                         Future<Optional<PluginInfo>> pluginInfo = fcpClient.loadPlugin().fromFreenet(KEY).execute();
 962                                         connectAndAssert(() -> createMatcher("freenet", KEY));
 963                                         replyWithPluginInfo();
 964                                         verifyPluginInfo(pluginInfo);
 965                                 }
 966
 967                                 private Matcher<List<String>> createMatcher(String urlType, String url) {
 968                                         return matchesFcpMessage(
 969                                                 "LoadPlugin",
 970                                                 "Identifier=" + identifier,
 971                                                 "PluginURL=" + url,
 972                                                 "URLType=" + urlType
 973                                         );
 974                                 }
 975
 976                         }
 977
 978                         public class Failed {
 979
 980                                 @Test
 981                                 public void failedLoad() throws ExecutionException, InterruptedException, IOException {
 982                                         Future<Optional<PluginInfo>> pluginInfo =
 983                                                 fcpClient.loadPlugin().officialFromFreenet("superPlugin").execute();
 984                                         connectAndAssert(() -> matchesFcpMessage("LoadPlugin"));
 985                                         replyWithProtocolError();
 986                                         assertThat(pluginInfo.get().isPresent(), is(false));
 987                                 }
 988
 989                         }
 990
 991                 }
 992
 993                 public class ReloadPlugin {
 994
 995                         @Test
 996                         public void reloadingPluginWorks() throws InterruptedException, ExecutionException, IOException {
 997                                 Future<Optional<PluginInfo>> pluginInfo = fcpClient.reloadPlugin().plugin(CLASS_NAME).execute();
 998                                 connectAndAssert(this::matchReloadPluginMessage);
 999                                 replyWithPluginInfo();
1000                                 verifyPluginInfo(pluginInfo);
1001                         }
1002
1003                         @Test
1004                         public void reloadingPluginWithMaxWaitTimeWorks()
1005                         throws InterruptedException, ExecutionException, IOException {
1006                                 Future<Optional<PluginInfo>> pluginInfo =
1007                                         fcpClient.reloadPlugin().waitFor(1234).plugin(CLASS_NAME).execute();
1008                                 connectAndAssert(() -> allOf(matchReloadPluginMessage(), hasItem("MaxWaitTime=1234")));
1009                                 replyWithPluginInfo();
1010                                 verifyPluginInfo(pluginInfo);
1011                         }
1012
1013                         @Test
1014                         public void reloadingPluginWithPurgeWorks()
1015                         throws InterruptedException, ExecutionException, IOException {
1016                                 Future<Optional<PluginInfo>> pluginInfo =
1017                                         fcpClient.reloadPlugin().purge().plugin(CLASS_NAME).execute();
1018                                 connectAndAssert(() -> allOf(matchReloadPluginMessage(), hasItem("Purge=true")));
1019                                 replyWithPluginInfo();
1020                                 verifyPluginInfo(pluginInfo);
1021                         }
1022
1023                         @Test
1024                         public void reloadingPluginWithStoreWorks()
1025                         throws InterruptedException, ExecutionException, IOException {
1026                                 Future<Optional<PluginInfo>> pluginInfo =
1027                                         fcpClient.reloadPlugin().addToConfig().plugin(CLASS_NAME).execute();
1028                                 connectAndAssert(() -> allOf(matchReloadPluginMessage(), hasItem("Store=true")));
1029                                 replyWithPluginInfo();
1030                                 verifyPluginInfo(pluginInfo);
1031                         }
1032
1033                         @Test
1034                         public void protocolErrorIsRecognizedAsFailure()
1035                         throws InterruptedException, ExecutionException, IOException {
1036                                 Future<Optional<PluginInfo>> pluginInfo = fcpClient.reloadPlugin().plugin(CLASS_NAME).execute();
1037                                 connectAndAssert(() -> matchReloadPluginMessage());
1038                                 replyWithProtocolError();
1039                                 assertThat(pluginInfo.get().isPresent(), is(false));
1040                         }
1041
1042                         private Matcher<List<String>> matchReloadPluginMessage() {
1043                                 return matchesFcpMessage(
1044                                         "ReloadPlugin",
1045                                         "Identifier=" + identifier,
1046                                         "PluginName=" + CLASS_NAME
1047                                 );
1048                         }
1049
1050                 }
1051
1052                 public class RemovePlugin {
1053
1054                         @Test
1055                         public void removingPluginWorks() throws InterruptedException, ExecutionException, IOException {
1056                                 Future<Boolean> pluginRemoved = fcpClient.removePlugin().plugin(CLASS_NAME).execute();
1057                                 connectAndAssert(this::matchPluginRemovedMessage);
1058                                 replyWithPluginRemoved();
1059                                 assertThat(pluginRemoved.get(), is(true));
1060                         }
1061
1062                         @Test
1063                         public void removingPluginWithMaxWaitTimeWorks()
1064                         throws InterruptedException, ExecutionException, IOException {
1065                                 Future<Boolean> pluginRemoved = fcpClient.removePlugin().waitFor(1234).plugin(CLASS_NAME).execute();
1066                                 connectAndAssert(() -> allOf(matchPluginRemovedMessage(), hasItem("MaxWaitTime=1234")));
1067                                 replyWithPluginRemoved();
1068                                 assertThat(pluginRemoved.get(), is(true));
1069                         }
1070
1071                         @Test
1072                         public void removingPluginWithPurgeWorks()
1073                         throws InterruptedException, ExecutionException, IOException {
1074                                 Future<Boolean> pluginRemoved = fcpClient.removePlugin().purge().plugin(CLASS_NAME).execute();
1075                                 connectAndAssert(() -> allOf(matchPluginRemovedMessage(), hasItem("Purge=true")));
1076                                 replyWithPluginRemoved();
1077                                 assertThat(pluginRemoved.get(), is(true));
1078                         }
1079
1080                         private void replyWithPluginRemoved() throws IOException {
1081                                 fcpServer.writeLine(
1082                                         "PluginRemoved",
1083                                         "Identifier=" + identifier,
1084                                         "PluginName=" + CLASS_NAME,
1085                                         "EndMessage"
1086                                 );
1087                         }
1088
1089                         private Matcher<List<String>> matchPluginRemovedMessage() {
1090                                 return matchesFcpMessage(
1091                                         "RemovePlugin",
1092                                         "Identifier=" + identifier,
1093                                         "PluginName=" + CLASS_NAME
1094                                 );
1095                         }
1096
1097                 }
1098
1099                 public class GetPluginInfo {
1100
1101                         @Test
1102                         public void gettingPluginInfoWorks() throws InterruptedException, ExecutionException, IOException {
1103                                 Future<Optional<PluginInfo>> pluginInfo = fcpClient.getPluginInfo().plugin(CLASS_NAME).execute();
1104                                 connectAndAssert(this::matchGetPluginInfoMessage);
1105                                 replyWithPluginInfo();
1106                                 verifyPluginInfo(pluginInfo);
1107                         }
1108
1109                         @Test
1110                         public void gettingPluginInfoWithDetailsWorks()
1111                         throws InterruptedException, ExecutionException, IOException {
1112                                 Future<Optional<PluginInfo>> pluginInfo =
1113                                         fcpClient.getPluginInfo().detailed().plugin(CLASS_NAME).execute();
1114                                 connectAndAssert(() -> allOf(matchGetPluginInfoMessage(), hasItem("Detailed=true")));
1115                                 replyWithPluginInfo();
1116                                 verifyPluginInfo(pluginInfo);
1117                         }
1118
1119                         @Test
1120                         public void protocolErrorIsRecognizedAsFailure()
1121                         throws InterruptedException, ExecutionException, IOException {
1122                                 Future<Optional<PluginInfo>> pluginInfo =
1123                                         fcpClient.getPluginInfo().detailed().plugin(CLASS_NAME).execute();
1124                                 connectAndAssert(() -> allOf(matchGetPluginInfoMessage(), hasItem("Detailed=true")));
1125                                 replyWithProtocolError();
1126                                 assertThat(pluginInfo.get(), is(Optional.empty()));
1127                         }
1128
1129                         private Matcher<List<String>> matchGetPluginInfoMessage() {
1130                                 return matchesFcpMessage(
1131                                         "GetPluginInfo",
1132                                         "Identifier=" + identifier,
1133                                         "PluginName=" + CLASS_NAME
1134                                 );
1135                         }
1136
1137                 }
1138
1139         }
1140
1141         public class UskSubscriptionCommands {
1142
1143                 private static final String URI = "USK@some,uri/file.txt";
1144
1145                 @Test
1146                 public void subscriptionWorks() throws InterruptedException, ExecutionException, IOException {
1147                         Future<Optional<UskSubscription>> uskSubscription = fcpClient.subscribeUsk().uri(URI).execute();
1148                         connectAndAssert(() -> matchesFcpMessage("SubscribeUSK", "URI=" + URI));
1149                         replyWithSubscribed();
1150                         assertThat(uskSubscription.get().get().getUri(), is(URI));
1151                         AtomicInteger edition = new AtomicInteger();
1152                         CountDownLatch updated = new CountDownLatch(2);
1153                         uskSubscription.get().get().onUpdate(e -> {
1154                                 edition.set(e);
1155                                 updated.countDown();
1156                         });
1157                         sendUpdateNotification(23);
1158                         sendUpdateNotification(24);
1159                         assertThat("updated in time", updated.await(5, TimeUnit.SECONDS), is(true));
1160                         assertThat(edition.get(), is(24));
1161                 }
1162
1163                 @Test
1164                 public void subscriptionUpdatesMultipleTimes() throws InterruptedException, ExecutionException, IOException {
1165                         Future<Optional<UskSubscription>> uskSubscription = fcpClient.subscribeUsk().uri(URI).execute();
1166                         connectAndAssert(() -> matchesFcpMessage("SubscribeUSK", "URI=" + URI));
1167                         replyWithSubscribed();
1168                         assertThat(uskSubscription.get().get().getUri(), is(URI));
1169                         AtomicInteger edition = new AtomicInteger();
1170                         CountDownLatch updated = new CountDownLatch(2);
1171                         uskSubscription.get().get().onUpdate(e -> {
1172                                 edition.set(e);
1173                                 updated.countDown();
1174                         });
1175                         uskSubscription.get().get().onUpdate(e -> updated.countDown());
1176                         sendUpdateNotification(23);
1177                         assertThat("updated in time", updated.await(5, TimeUnit.SECONDS), is(true));
1178                         assertThat(edition.get(), is(23));
1179                 }
1180
1181                 @Test
1182                 public void subscriptionCanBeCancelled() throws InterruptedException, ExecutionException, IOException {
1183                         Future<Optional<UskSubscription>> uskSubscription = fcpClient.subscribeUsk().uri(URI).execute();
1184                         connectAndAssert(() -> matchesFcpMessage("SubscribeUSK", "URI=" + URI));
1185                         replyWithSubscribed();
1186                         assertThat(uskSubscription.get().get().getUri(), is(URI));
1187                         AtomicBoolean updated = new AtomicBoolean();
1188                         uskSubscription.get().get().onUpdate(e -> updated.set(true));
1189                         uskSubscription.get().get().cancel();
1190                         readMessage(() -> matchesFcpMessage("UnsubscribeUSK", "Identifier=" + identifier));
1191                         sendUpdateNotification(23);
1192                         assertThat(updated.get(), is(false));
1193                 }
1194
1195                 private void replyWithSubscribed() throws IOException {
1196                         fcpServer.writeLine(
1197                                 "SubscribedUSK",
1198                                 "Identifier=" + identifier,
1199                                 "URI=" + URI,
1200                                 "DontPoll=false",
1201                                 "EndMessage"
1202                         );
1203                 }
1204
1205                 private void sendUpdateNotification(int edition, String... additionalLines) throws IOException {
1206                         fcpServer.writeLine(
1207                                 "SubscribedUSKUpdate",
1208                                 "Identifier=" + identifier,
1209                                 "URI=" + URI,
1210                                 "Edition=" + edition
1211                         );
1212                         fcpServer.writeLine(additionalLines);
1213                         fcpServer.writeLine("EndMessage");
1214                 }
1215
1216         }
1217
1218         public class ClientGet {
1219
1220                 @Test
1221                 public void works() throws InterruptedException, ExecutionException, IOException {
1222                         Future<Optional<Data>> dataFuture = fcpClient.clientGet().uri("KSK@foo.txt").execute();
1223                         connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt", "ReturnType=direct"));
1224                         replyWithAllData("not-test", "Hello World", "text/plain;charset=latin-9");
1225                         replyWithAllData(identifier, "Hello", "text/plain;charset=utf-8");
1226                         Optional<Data> data = dataFuture.get();
1227                         verifyData(data);
1228                 }
1229
1230                 @Test
1231                 public void getFailedIsRecognized() throws InterruptedException, ExecutionException, IOException {
1232                         Future<Optional<Data>> dataFuture = fcpClient.clientGet().uri("KSK@foo.txt").execute();
1233                         connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt"));
1234                         replyWithGetFailed("not-test");
1235                         replyWithGetFailed(identifier);
1236                         Optional<Data> data = dataFuture.get();
1237                         assertThat(data.isPresent(), is(false));
1238                 }
1239
1240                 @Test
1241                 public void getFailedForDifferentIdentifierIsIgnored()
1242                 throws InterruptedException, ExecutionException, IOException {
1243                         Future<Optional<Data>> dataFuture = fcpClient.clientGet().uri("KSK@foo.txt").execute();
1244                         connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt"));
1245                         replyWithGetFailed("not-test");
1246                         replyWithAllData(identifier, "Hello", "text/plain;charset=utf-8");
1247                         Optional<Data> data = dataFuture.get();
1248                         verifyData(data);
1249                 }
1250
1251                 @Test(expected = ExecutionException.class)
1252                 public void connectionClosedIsRecognized() throws InterruptedException, ExecutionException, IOException {
1253                         Future<Optional<Data>> dataFuture = fcpClient.clientGet().uri("KSK@foo.txt").execute();
1254                         connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt"));
1255                         fcpServer.close();
1256                         dataFuture.get();
1257                 }
1258
1259                 @Test
1260                 public void withIgnoreData() throws InterruptedException, ExecutionException, IOException {
1261                         fcpClient.clientGet().ignoreDataStore().uri("KSK@foo.txt").execute();
1262                         connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt", "IgnoreDS=true"));
1263                 }
1264
1265                 @Test
1266                 public void withDataStoreOnly() throws InterruptedException, ExecutionException, IOException {
1267                         fcpClient.clientGet().dataStoreOnly().uri("KSK@foo.txt").execute();
1268                         connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt", "DSonly=true"));
1269                 }
1270
1271                 @Test
1272                 public void clientGetWithMaxSizeSettingSendsCorrectCommands()
1273                 throws InterruptedException, ExecutionException, IOException {
1274                         fcpClient.clientGet().maxSize(1048576).uri("KSK@foo.txt").execute();
1275                         connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt", "MaxSize=1048576"));
1276                 }
1277
1278                 @Test
1279                 public void clientGetWithPrioritySettingSendsCorrectCommands()
1280                 throws InterruptedException, ExecutionException, IOException {
1281                         fcpClient.clientGet().priority(Priority.interactive).uri("KSK@foo.txt").execute();
1282                         connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt", "PriorityClass=1"));
1283                 }
1284
1285                 @Test
1286                 public void clientGetWithRealTimeSettingSendsCorrectCommands()
1287                 throws InterruptedException, ExecutionException, IOException {
1288                         fcpClient.clientGet().realTime().uri("KSK@foo.txt").execute();
1289                         connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt", "RealTimeFlag=true"));
1290                 }
1291
1292                 @Test
1293                 public void clientGetWithGlobalSettingSendsCorrectCommands()
1294                 throws InterruptedException, ExecutionException, IOException {
1295                         fcpClient.clientGet().global().uri("KSK@foo.txt").execute();
1296                         connectAndAssert(() -> matchesFcpMessage("ClientGet", "URI=KSK@foo.txt", "Global=true"));
1297                 }
1298
1299                 private void replyWithGetFailed(String identifier) throws IOException {
1300                         fcpServer.writeLine(
1301                                 "GetFailed",
1302                                 "Identifier=" + identifier,
1303                                 "Code=3",
1304                                 "EndMessage"
1305                         );
1306                 }
1307
1308                 private void replyWithAllData(String identifier, String text, String contentType) throws IOException {
1309                         fcpServer.writeLine(
1310                                 "AllData",
1311                                 "Identifier=" + identifier,
1312                                 "DataLength=" + (text.length() + 1),
1313                                 "StartupTime=1435610539000",
1314                                 "CompletionTime=1435610540000",
1315                                 "Metadata.ContentType=" + contentType,
1316                                 "Data",
1317                                 text
1318                         );
1319                 }
1320
1321                 private void verifyData(Optional<Data> data) throws IOException {
1322                         assertThat(data.get().getMimeType(), is("text/plain;charset=utf-8"));
1323                         assertThat(data.get().size(), is(6L));
1324                         assertThat(ByteStreams.toByteArray(data.get().getInputStream()),
1325                                 is("Hello\n".getBytes(StandardCharsets.UTF_8)));
1326                 }
1327
1328         }
1329
1330         public class ClientPut {
1331
1332                 @Test
1333                 public void sendsCorrectCommand() throws IOException, ExecutionException, InterruptedException {
1334                         fcpClient.clientPut()
1335                                 .from(new ByteArrayInputStream("Hello\n".getBytes()))
1336                                 .length(6)
1337                                 .uri("KSK@foo.txt")
1338                                 .execute();
1339                         connectNode();
1340                         readMessage("Hello", this::matchesDirectClientPut);
1341                 }
1342
1343                 @Test
1344                 public void succeedsOnCorrectIdentifier() throws InterruptedException, ExecutionException, IOException {
1345                         Future<Optional<Key>> key = fcpClient.clientPut()
1346                                 .from(new ByteArrayInputStream("Hello\n".getBytes()))
1347                                 .length(6)
1348                                 .uri("KSK@foo.txt")
1349                                 .execute();
1350                         connectNode();
1351                         readMessage("Hello", this::matchesDirectClientPut);
1352                         replyWithPutFailed("not-the-right-one");
1353                         replyWithPutSuccessful(identifier);
1354                         assertThat(key.get().get().getKey(), is("KSK@foo.txt"));
1355                 }
1356
1357                 @Test
1358                 public void failsOnCorrectIdentifier() throws InterruptedException, ExecutionException, IOException {
1359                         Future<Optional<Key>> key = fcpClient.clientPut()
1360                                 .from(new ByteArrayInputStream("Hello\n".getBytes()))
1361                                 .length(6)
1362                                 .uri("KSK@foo.txt")
1363                                 .execute();
1364                         connectNode();
1365                         readMessage("Hello", this::matchesDirectClientPut);
1366                         replyWithPutSuccessful("not-the-right-one");
1367                         replyWithPutFailed(identifier);
1368                         assertThat(key.get().isPresent(), is(false));
1369                 }
1370
1371                 @Test
1372                 public void renameIsSentCorrectly() throws InterruptedException, ExecutionException, IOException {
1373                         fcpClient.clientPut()
1374                                 .named("otherName.txt")
1375                                 .from(new ByteArrayInputStream("Hello\n".getBytes()))
1376                                 .length(6)
1377                                 .uri("KSK@foo.txt")
1378                                 .execute();
1379                         connectNode();
1380                         readMessage("Hello", () -> allOf(
1381                                 hasHead("ClientPut"),
1382                                 hasParameters(1, 2, "TargetFilename=otherName.txt", "UploadFrom=direct", "DataLength=6",
1383                                         "URI=KSK@foo.txt"),
1384                                 hasTail("EndMessage", "Hello")
1385                         ));
1386                 }
1387
1388                 @Test
1389                 public void redirectIsSentCorrecly() throws IOException, ExecutionException, InterruptedException {
1390                         fcpClient.clientPut().redirectTo("KSK@bar.txt").uri("KSK@foo.txt").execute();
1391                         connectAndAssert(() ->
1392                                 matchesFcpMessage("ClientPut", "UploadFrom=redirect", "URI=KSK@foo.txt", "TargetURI=KSK@bar.txt"));
1393                 }
1394
1395                 @Test
1396                 public void withFileIsSentCorrectly() throws InterruptedException, ExecutionException, IOException {
1397                         fcpClient.clientPut().from(new File("/tmp/data.txt")).uri("KSK@foo.txt").execute();
1398                         connectAndAssert(() ->
1399                                 matchesFcpMessage("ClientPut", "UploadFrom=disk", "URI=KSK@foo.txt", "Filename=/tmp/data.txt"));
1400                 }
1401
1402                 public class DDA {
1403
1404                         private final File ddaFile;
1405                         private final File fileToUpload;
1406
1407                         public DDA() throws IOException {
1408                                 ddaFile = createDdaFile();
1409                                 fileToUpload = new File(ddaFile.getParent(), "test.dat");
1410                         }
1411
1412                         private Matcher<List<String>> matchesFileClientPut(File file) {
1413                                 return matchesFcpMessage("ClientPut", "UploadFrom=disk", "URI=KSK@foo.txt", "Filename=" + file);
1414                         }
1415
1416                         @Test
1417                         public void completeDda() throws IOException, ExecutionException, InterruptedException {
1418                                 fcpClient.clientPut().from(fileToUpload).uri("KSK@foo.txt").execute();
1419                                 connectAndAssert(() -> matchesFileClientPut(fileToUpload));
1420                                 sendDdaRequired(identifier);
1421                                 readMessage(() -> matchesTestDDARequest(ddaFile));
1422                                 sendTestDDAReply(ddaFile.getParent(), ddaFile);
1423                                 readMessage(() -> matchesTestDDAResponse(ddaFile));
1424                                 writeTestDDAComplete(ddaFile);
1425                                 readMessage(() -> matchesFileClientPut(fileToUpload));
1426                         }
1427
1428                         @Test
1429                         public void ignoreOtherDda() throws IOException, ExecutionException, InterruptedException {
1430                                 fcpClient.clientPut().from(fileToUpload).uri("KSK@foo.txt").execute();
1431                                 connectAndAssert(() -> matchesFileClientPut(fileToUpload));
1432                                 sendDdaRequired(identifier);
1433                                 readMessage(() -> matchesTestDDARequest(ddaFile));
1434                                 sendTestDDAReply("/some-other-directory", ddaFile);
1435                                 sendTestDDAReply(ddaFile.getParent(), ddaFile);
1436                                 readMessage(() -> matchesTestDDAResponse(ddaFile));
1437                         }
1438
1439                         @Test
1440                         public void sendResponseIfFileUnreadable() throws IOException, ExecutionException, InterruptedException {
1441                                 fcpClient.clientPut().from(fileToUpload).uri("KSK@foo.txt").execute();
1442                                 connectAndAssert(() -> matchesFileClientPut(fileToUpload));
1443                                 sendDdaRequired(identifier);
1444                                 readMessage(() -> matchesTestDDARequest(ddaFile));
1445                                 sendTestDDAReply(ddaFile.getParent(), new File(ddaFile + ".foo"));
1446                                 readMessage(this::matchesFailedToReadResponse);
1447                         }
1448
1449                         @Test
1450                         public void clientPutDoesNotResendOriginalClientPutOnTestDDACompleteWithWrongDirectory()
1451                         throws IOException, ExecutionException, InterruptedException {
1452                                 fcpClient.clientPut().from(fileToUpload).uri("KSK@foo.txt").execute();
1453                                 connectNode();
1454                                 List<String> lines = fcpServer.collectUntil(is("EndMessage"));
1455                                 String identifier = extractIdentifier(lines);
1456                                 fcpServer.writeLine(
1457                                         "TestDDAComplete",
1458                                         "Directory=/some-other-directory",
1459                                         "EndMessage"
1460                                 );
1461                                 sendDdaRequired(identifier);
1462                                 lines = fcpServer.collectUntil(is("EndMessage"));
1463                                 assertThat(lines, matchesFcpMessage(
1464                                         "TestDDARequest",
1465                                         "Directory=" + ddaFile.getParent(),
1466                                         "WantReadDirectory=true",
1467                                         "WantWriteDirectory=false"
1468                                 ));
1469                         }
1470
1471                         private Matcher<List<String>> matchesFailedToReadResponse() {
1472                                 return matchesFcpMessage(
1473                                         "TestDDAResponse",
1474                                         "Directory=" + ddaFile.getParent(),
1475                                         "ReadContent=failed-to-read"
1476                                 );
1477                         }
1478
1479                         private void writeTestDDAComplete(File tempFile) throws IOException {
1480                                 fcpServer.writeLine(
1481                                         "TestDDAComplete",
1482                                         "Directory=" + tempFile.getParent(),
1483                                         "ReadDirectoryAllowed=true",
1484                                         "EndMessage"
1485                                 );
1486                         }
1487
1488                         private Matcher<List<String>> matchesTestDDAResponse(File tempFile) {
1489                                 return matchesFcpMessage(
1490                                         "TestDDAResponse",
1491                                         "Directory=" + tempFile.getParent(),
1492                                         "ReadContent=test-content"
1493                                 );
1494                         }
1495
1496                         private void sendTestDDAReply(String directory, File tempFile) throws IOException {
1497                                 fcpServer.writeLine(
1498                                         "TestDDAReply",
1499                                         "Directory=" + directory,
1500                                         "ReadFilename=" + tempFile,
1501                                         "EndMessage"
1502                                 );
1503                         }
1504
1505                         private Matcher<List<String>> matchesTestDDARequest(File tempFile) {
1506                                 return matchesFcpMessage(
1507                                         "TestDDARequest",
1508                                         "Directory=" + tempFile.getParent(),
1509                                         "WantReadDirectory=true",
1510                                         "WantWriteDirectory=false"
1511                                 );
1512                         }
1513
1514                         private void sendDdaRequired(String identifier) throws IOException {
1515                                 fcpServer.writeLine(
1516                                         "ProtocolError",
1517                                         "Identifier=" + identifier,
1518                                         "Code=25",
1519                                         "EndMessage"
1520                                 );
1521                         }
1522
1523                 }
1524
1525                 private void replyWithPutSuccessful(String identifier) throws IOException {
1526                         fcpServer.writeLine(
1527                                 "PutSuccessful",
1528                                 "URI=KSK@foo.txt",
1529                                 "Identifier=" + identifier,
1530                                 "EndMessage"
1531                         );
1532                 }
1533
1534                 private void replyWithPutFailed(String identifier) throws IOException {
1535                         fcpServer.writeLine(
1536                                 "PutFailed",
1537                                 "Identifier=" + identifier,
1538                                 "EndMessage"
1539                         );
1540                 }
1541
1542                 private Matcher<List<String>> matchesDirectClientPut() {
1543                         return allOf(
1544                                 hasHead("ClientPut"),
1545                                 hasParameters(1, 2, "UploadFrom=direct", "DataLength=6", "URI=KSK@foo.txt"),
1546                                 hasTail("EndMessage", "Hello")
1547                         );
1548                 }
1549
1550                 private File createDdaFile() throws IOException {
1551                         File tempFile = File.createTempFile("test-dda-", ".dat");
1552                         tempFile.deleteOnExit();
1553                         Files.write("test-content", tempFile, StandardCharsets.UTF_8);
1554                         return tempFile;
1555                 }
1556
1557                 @Test
1558                 public void clientPutDoesNotReactToProtocolErrorForDifferentIdentifier()
1559                 throws InterruptedException, ExecutionException, IOException {
1560                         Future<Optional<Key>> key = fcpClient.clientPut().from(new File("/tmp/data.txt")).uri("KSK@foo.txt").execute();
1561                         connectNode();
1562                         List<String> lines = fcpServer.collectUntil(is("EndMessage"));
1563                         String identifier = extractIdentifier(lines);
1564                         fcpServer.writeLine(
1565                                 "ProtocolError",
1566                                 "Identifier=not-the-right-one",
1567                                 "Code=25",
1568                                 "EndMessage"
1569                         );
1570                         fcpServer.writeLine(
1571                                 "PutSuccessful",
1572                                 "Identifier=" + identifier,
1573                                 "URI=KSK@foo.txt",
1574                                 "EndMessage"
1575                         );
1576                         assertThat(key.get().get().getKey(), is("KSK@foo.txt"));
1577                 }
1578
1579                 @Test
1580                 public void clientPutAbortsOnProtocolErrorOtherThan25()
1581                 throws InterruptedException, ExecutionException, IOException {
1582                         Future<Optional<Key>> key = fcpClient.clientPut().from(new File("/tmp/data.txt")).uri("KSK@foo.txt").execute();
1583                         connectNode();
1584                         List<String> lines = fcpServer.collectUntil(is("EndMessage"));
1585                         String identifier = extractIdentifier(lines);
1586                         fcpServer.writeLine(
1587                                 "ProtocolError",
1588                                 "Identifier=" + identifier,
1589                                 "Code=1",
1590                                 "EndMessage"
1591                         );
1592                         assertThat(key.get().isPresent(), is(false));
1593                 }
1594
1595                 @Test
1596                 public void clientPutSendsNotificationsForGeneratedKeys()
1597                 throws InterruptedException, ExecutionException, IOException {
1598                         List<String> generatedKeys = new CopyOnWriteArrayList<>();
1599                         Future<Optional<Key>> key = fcpClient.clientPut()
1600                                 .onKeyGenerated(generatedKeys::add)
1601                                 .from(new ByteArrayInputStream("Hello\n".getBytes()))
1602                                 .length(6)
1603                                 .uri("KSK@foo.txt")
1604                                 .execute();
1605                         connectNode();
1606                         List<String> lines = fcpServer.collectUntil(is("Hello"));
1607                         String identifier = extractIdentifier(lines);
1608                         fcpServer.writeLine(
1609                                 "URIGenerated",
1610                                 "Identifier=" + identifier,
1611                                 "URI=KSK@foo.txt",
1612                                 "EndMessage"
1613                         );
1614                         replyWithPutSuccessful(identifier);
1615                         assertThat(key.get().get().getKey(), is("KSK@foo.txt"));
1616                         assertThat(generatedKeys, contains("KSK@foo.txt"));
1617                 }
1618
1619         }
1620
1621         public class ConfigCommand {
1622
1623                 public class GetConfig {
1624
1625                         @Test
1626                         public void defaultFcpClientCanGetConfigWithoutDetails()
1627                         throws InterruptedException, ExecutionException, IOException {
1628                                 Future<ConfigData> configData = fcpClient.getConfig().execute();
1629                                 connectAndAssert(() -> matchesFcpMessage("GetConfig", "Identifier=" + identifier));
1630                                 replyWithConfigData();
1631                                 assertThat(configData.get(), notNullValue());
1632                         }
1633
1634                         @Test
1635                         public void defaultFcpClientCanGetConfigWithCurrent()
1636                         throws InterruptedException, ExecutionException, IOException {
1637                                 Future<ConfigData> configData = fcpClient.getConfig().withCurrent().execute();
1638                                 connectAndAssert(() -> matchesGetConfigWithAdditionalParameter("WithCurrent"));
1639                                 replyWithConfigData("current.foo=bar");
1640                                 assertThat(configData.get().getCurrent("foo"), is("bar"));
1641                         }
1642
1643                         @Test
1644                         public void defaultFcpClientCanGetConfigWithDefaults()
1645                         throws InterruptedException, ExecutionException, IOException {
1646                                 Future<ConfigData> configData = fcpClient.getConfig().withDefaults().execute();
1647                                 connectAndAssert(() -> matchesGetConfigWithAdditionalParameter("WithDefaults"));
1648                                 replyWithConfigData("default.foo=bar");
1649                                 assertThat(configData.get().getDefault("foo"), is("bar"));
1650                         }
1651
1652                         @Test
1653                         public void defaultFcpClientCanGetConfigWithSortOrder()
1654                         throws InterruptedException, ExecutionException, IOException {
1655                                 Future<ConfigData> configData = fcpClient.getConfig().withSortOrder().execute();
1656                                 connectAndAssert(() -> matchesGetConfigWithAdditionalParameter("WithSortOrder"));
1657                                 replyWithConfigData("sortOrder.foo=17");
1658                                 assertThat(configData.get().getSortOrder("foo"), is(17));
1659                         }
1660
1661                         @Test
1662                         public void defaultFcpClientCanGetConfigWithExpertFlag()
1663                         throws InterruptedException, ExecutionException, IOException {
1664                                 Future<ConfigData> configData = fcpClient.getConfig().withExpertFlag().execute();
1665                                 connectAndAssert(() -> matchesGetConfigWithAdditionalParameter("WithExpertFlag"));
1666                                 replyWithConfigData("expertFlag.foo=true");
1667                                 assertThat(configData.get().getExpertFlag("foo"), is(true));
1668                         }
1669
1670                         @Test
1671                         public void defaultFcpClientCanGetConfigWithForceWriteFlag()
1672                         throws InterruptedException, ExecutionException, IOException {
1673                                 Future<ConfigData> configData = fcpClient.getConfig().withForceWriteFlag().execute();
1674                                 connectAndAssert(() -> matchesGetConfigWithAdditionalParameter("WithForceWriteFlag"));
1675                                 replyWithConfigData("forceWriteFlag.foo=true");
1676                                 assertThat(configData.get().getForceWriteFlag("foo"), is(true));
1677                         }
1678
1679                         @Test
1680                         public void defaultFcpClientCanGetConfigWithShortDescription()
1681                         throws InterruptedException, ExecutionException, IOException {
1682                                 Future<ConfigData> configData = fcpClient.getConfig().withShortDescription().execute();
1683                                 connectAndAssert(() -> matchesGetConfigWithAdditionalParameter("WithShortDescription"));
1684                                 replyWithConfigData("shortDescription.foo=bar");
1685                                 assertThat(configData.get().getShortDescription("foo"), is("bar"));
1686                         }
1687
1688                         @Test
1689                         public void defaultFcpClientCanGetConfigWithLongDescription()
1690                         throws InterruptedException, ExecutionException, IOException {
1691                                 Future<ConfigData> configData = fcpClient.getConfig().withLongDescription().execute();
1692                                 connectAndAssert(() -> matchesGetConfigWithAdditionalParameter("WithLongDescription"));
1693                                 replyWithConfigData("longDescription.foo=bar");
1694                                 assertThat(configData.get().getLongDescription("foo"), is("bar"));
1695                         }
1696
1697                         @Test
1698                         public void defaultFcpClientCanGetConfigWithDataTypes()
1699                         throws InterruptedException, ExecutionException, IOException {
1700                                 Future<ConfigData> configData = fcpClient.getConfig().withDataTypes().execute();
1701                                 connectAndAssert(() -> matchesGetConfigWithAdditionalParameter("WithDataTypes"));
1702                                 replyWithConfigData("dataType.foo=number");
1703                                 assertThat(configData.get().getDataType("foo"), is("number"));
1704                         }
1705
1706                         private Matcher<List<String>> matchesGetConfigWithAdditionalParameter(String additionalParameter) {
1707                                 return matchesFcpMessage(
1708                                         "GetConfig",
1709                                         "Identifier=" + identifier,
1710                                         additionalParameter + "=true"
1711                                 );
1712                         }
1713
1714                 }
1715
1716                 public class ModifyConfig {
1717
1718                         @Test
1719                         public void defaultFcpClientCanModifyConfigData()
1720                         throws InterruptedException, ExecutionException, IOException {
1721                                 Future<ConfigData> newConfigData = fcpClient.modifyConfig().set("foo.bar").to("baz").execute();
1722                                 connectAndAssert(() -> matchesFcpMessage(
1723                                         "ModifyConfig",
1724                                         "Identifier=" + identifier,
1725                                         "foo.bar=baz"
1726                                 ));
1727                                 replyWithConfigData("current.foo.bar=baz");
1728                                 assertThat(newConfigData.get().getCurrent("foo.bar"), is("baz"));
1729                         }
1730
1731                 }
1732
1733                 private void replyWithConfigData(String... additionalLines) throws IOException {
1734                         fcpServer.writeLine("ConfigData", "Identifier=" + identifier);
1735                         fcpServer.writeLine(additionalLines);
1736                         fcpServer.writeLine("EndMessage");
1737                 }
1738
1739         }
1740
1741         public class NodeInformation {
1742
1743                 @Test
1744                 public void defaultFcpClientCanGetNodeInformation() throws InterruptedException, ExecutionException, IOException {
1745                         Future<NodeData> nodeData = fcpClient.getNode().execute();
1746                         connectAndAssert(() -> matchesGetNode(false, false, false));
1747                         replyWithNodeData();
1748                         assertThat(nodeData.get(), notNullValue());
1749                         assertThat(nodeData.get().getNodeRef().isOpennet(), is(false));
1750                 }
1751
1752                 @Test
1753                 public void defaultFcpClientCanGetNodeInformationWithOpennetRef()
1754                 throws InterruptedException, ExecutionException, IOException {
1755                         Future<NodeData> nodeData = fcpClient.getNode().opennetRef().execute();
1756                         connectAndAssert(() -> matchesGetNode(true, false, false));
1757                         replyWithNodeData("opennet=true");
1758                         assertThat(nodeData.get().getVersion().toString(), is("Fred,0.7,1.0,1466"));
1759                         assertThat(nodeData.get().getNodeRef().isOpennet(), is(true));
1760                 }
1761
1762                 @Test
1763                 public void defaultFcpClientCanGetNodeInformationWithPrivateData()
1764                 throws InterruptedException, ExecutionException, IOException {
1765                         Future<NodeData> nodeData = fcpClient.getNode().includePrivate().execute();
1766                         connectAndAssert(() -> matchesGetNode(false, true, false));
1767                         replyWithNodeData("ark.privURI=SSK@XdHMiRl");
1768                         assertThat(nodeData.get().getARK().getPrivateURI(), is("SSK@XdHMiRl"));
1769                 }
1770
1771                 @Test
1772                 public void defaultFcpClientCanGetNodeInformationWithVolatileData()
1773                 throws InterruptedException, ExecutionException, IOException {
1774                         Future<NodeData> nodeData = fcpClient.getNode().includeVolatile().execute();
1775                         connectAndAssert(() -> matchesGetNode(false, false, true));
1776                         replyWithNodeData("volatile.freeJavaMemory=205706528");
1777                         assertThat(nodeData.get().getVolatile("freeJavaMemory"), is("205706528"));
1778                 }
1779
1780                 private Matcher<List<String>> matchesGetNode(boolean withOpennetRef, boolean withPrivate, boolean withVolatile) {
1781                         return matchesFcpMessage(
1782                                 "GetNode",
1783                                 "Identifier=" + identifier,
1784                                 "GiveOpennetRef=" + withOpennetRef,
1785                                 "WithPrivate=" + withPrivate,
1786                                 "WithVolatile=" + withVolatile
1787                         );
1788                 }
1789
1790                 private void replyWithNodeData(String... additionalLines) throws IOException {
1791                         fcpServer.writeLine(
1792                                 "NodeData",
1793                                 "Identifier=" + identifier,
1794                                 "ark.pubURI=SSK@3YEf.../ark",
1795                                 "ark.number=78",
1796                                 "auth.negTypes=2",
1797                                 "version=Fred,0.7,1.0,1466",
1798                                 "lastGoodVersion=Fred,0.7,1.0,1466"
1799                         );
1800                         fcpServer.writeLine(additionalLines);
1801                         fcpServer.writeLine("EndMessage");
1802                 }
1803
1804         }
1805
1806 }