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