pyrepo: test with testDataStoreBasic

.github/workflows/build.yml

@@ -38,12 +38,16 @@ jobs:
       - uses: actions/setup-node@v4
         with:
           node-version: ${{ matrix.node }}
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.12'
       - name: Install dependencies
         run: |
           echo "deb [arch=amd64 trusted=yes] https://nfd-nightly-apt.ndn.today/ubuntu jammy main" \
             | sudo tee /etc/apt/sources.list.d/nfd-nightly.list
           sudo apt-get update
           sudo apt-get install --no-install-recommends ndnsec
+          pip install git+https://github.com/UCLA-IRL/ndn-python-repo@2dcd229a4cb81927a52e8a8f1d963c55ee939ffa
       - run: corepack pnpm install
         env:
           PUPPETEER_SKIP_DOWNLOAD: true

pkg/pyrepo/README.md

@@ -3,24 +3,22 @@
 This package is part of [NDNts](https://yoursunny.com/p/NDNts/), Named Data Networking libraries for the modern web.
 
 This package allows inserting and deleting Data in [ndn-python-repo](https://github.com/UCLA-IRL/ndn-python-repo).
-This implementation is compatible with ndn-python-repo `dafd23dc` (2024-02-13).
+This implementation is compatible with ndn-python-repo `2dcd229a` (2024-05-09).
 To install and start the specified version, run:
 
 ```bash
 # create Python virtual environment
-python3 -m venv pyrepo-venv
-cd pyrepo-venv
-source ./bin/activate
+python3 -m venv ~/pyrepo-venv
+source ~/pyrepo-venv/bin/activate
 
 # install ndn-python-repo
-pip install git+https://github.com/UCLA-IRL/ndn-python-repo@dafd23dcc25bf9c130a110e37b66d6d1683a8212
+pip install git+https://github.com/UCLA-IRL/ndn-python-repo@2dcd229a4cb81927a52e8a8f1d963c55ee939ffa
 
 # run ndn-python-repo
-export NDN_CLIENT_TRANSPORT=unix:///run/nfd/nfd.sock
 ndn-python-repo
 ```
 
-`PyRepoClient` type is a client for [ndn-python-repo protocol](https://github.com/UCLA-IRL/ndn-python-repo/tree/dafd23dcc25bf9c130a110e37b66d6d1683a8212/docs/src/specification).
+`PyRepoClient` type is a client for [ndn-python-repo protocol](https://github.com/UCLA-IRL/ndn-python-repo/tree/2dcd229a4cb81927a52e8a8f1d963c55ee939ffa/docs/src/specification).
 `PyRepoStore` type implements a write-only subset of `DataStore` interfaces as defined in `@ndn/repo-api` package.
 
 ```ts

pkg/pyrepo/src/client.ts

@@ -1,6 +1,6 @@
 import { consume, ConsumerOptions } from "@ndn/endpoint";
 import { Segment } from "@ndn/naming-convention2";
-import { digestSigning, Interest, Name, SignedInterestPolicy } from "@ndn/packet";
+import { type Data, digestSigning, Interest, Name, SignedInterestPolicy } from "@ndn/packet";
 import { Decoder, Encoder } from "@ndn/tlv";
 import { assert, delay, randomJitter, sha256, toHex } from "@ndn/util";
 
@@ -106,12 +106,11 @@ export class PyRepoClient implements Disposable {
     const p = new CommandParam();
     p.objectParams.push(...objs);
     const request = Encoder.encode(p);
-    const requestDigest = await sha256(request);
-    const requestDigestHex = toHex(requestDigest);
 
     await this.publisher.publish(this.repoPrefix.append(verb.action), request);
     const checkParam = new StatQuery();
-    checkParam.requestDigest = requestDigest;
+    checkParam.requestDigest = await sha256(request);
+    const requestDigestHex = toHex(checkParam.requestDigest);
 
     const deadline = Date.now() + this.commandTimeout;
     while (Date.now() < deadline) {
@@ -123,10 +122,15 @@ export class PyRepoClient implements Disposable {
       checkSIP.update(checkInterest, this);
       await digestSigning.sign(checkInterest);
 
-      const checkData = await consume(checkInterest, {
-        ...this.checkOpts,
-        describe: `pyrepo-check(${this.repoPrefix} ${requestDigestHex})`,
-      });
+      let checkData: Data;
+      try {
+        checkData = await consume(checkInterest, {
+          ...this.checkOpts,
+          describe: `pyrepo-check(${this.repoPrefix} ${requestDigestHex})`,
+        });
+      } catch {
+        continue;
+      }
 
       const res = Decoder.decode(checkData.content, CommandRes);
       if (res.statusCode >= 400) {

pkg/pyrepo/src/store.ts

@@ -14,7 +14,8 @@ import { PyRepoClient } from "./client";
  * This DataStore is write-only. It can insert and delete data in ndn-python-repo.
  *
  * This DataStore does not have methods to read data. To read data in ndn-python-repo, send an
- * Interest to the network, and then ndn-python-repo is supposed to reply.
+ * Interest to the network, and then ndn-python-repo itself can reply.
+ * If you really need a readable DataStore, refer to {@link @ndn/repo-api!ReadFromNetwork}.
  */
 export class PyRepoStore implements Disposable, S.Insert, S.Delete {
   /** Construct with internal {@link PyRepoClient}. */

pkg/pyrepo/test-fixture/pyrepo.ts

@@ -28,7 +28,7 @@ export class PyRepo implements AsyncDisposable {
     name = Name.from(name);
     const dbFile = path.join(dir, "sqlite3.db");
     const confFile = path.join(dir, "repo.conf.json");
-    const ip = long2ip(0x7F000000 | Math.trunc(0x00FFFFFF * Math.random()));
+    const ip = long2ip(0x7F790000 | Math.trunc(0xFFFF * Math.random())); // 127.121.x.x
 
     using closers = new Closers();
     const nfd = await new FakeNfd(fw).open();
@@ -52,7 +52,7 @@ export class PyRepo implements AsyncDisposable {
       stdout: "inherit",
       stderr: "inherit",
       env: {
-        NDN_CLIENT_TRANSPORT: `tcp://127.0.0.1:${nfd.port}`,
+        NDN_CLIENT_TRANSPORT: `tcp://${ip}:${nfd.port}`,
         HOME: dir,
       },
     });
@@ -68,10 +68,10 @@ export class PyRepo implements AsyncDisposable {
 
   public async [Symbol.asyncDispose](): Promise<void> {
     this.p.kill("SIGQUIT");
-    try {
-      await this.p;
-    } catch {}
-    await this.nfd[Symbol.asyncDispose]();
+    await Promise.allSettled([
+      this.p,
+      this.nfd[Symbol.asyncDispose](),
+    ]);
   }
 }
 export namespace PyRepo {

pkg/pyrepo/tests/prps.t.ts

@@ -57,7 +57,7 @@ test("pubsub", { timeout: 10000, retry: 1 }, async () => {
     Promise.allSettled(pubPromises),
     (async () => {
       using closers = new Closers(sub0A, sub0B, sub1A);
-      await delay(3000);
+      await delay(4000);
     })(),
   ]);
 

pkg/pyrepo/tests/pyrepo.t.ts

@@ -4,27 +4,45 @@ import { consume, type ConsumerOptions } from "@ndn/endpoint";
 import { Forwarder } from "@ndn/fw";
 import { Segment } from "@ndn/naming-convention2";
 import { Data, digestSigning, Name } from "@ndn/packet";
+import { ReadFromNetwork } from "@ndn/repo-api";
+import { testDataStoreBasic } from "@ndn/repo-api/test-fixture/data-store";
+import { Closers, delay } from "@ndn/util";
 import { makeTmpDir } from "@ndn/util/test-fixture/tmp";
-import { expect, test } from "vitest";
+import { parallelMap } from "streaming-iterables";
+import { beforeEach, expect, test } from "vitest";
 
 import { PyRepoStore } from "..";
 import { PyRepo } from "../test-fixture/pyrepo";
 
-afterEach(Forwarder.deleteDefault);
+const closers = new Closers();
+let store: PyRepoStore;
+beforeEach(async () => {
+  const tmpDir = makeTmpDir();
+  closers.push(tmpDir);
+  const repo = await PyRepo.create("/myrepo", { dir: tmpDir.name });
+  closers.push(repo);
 
-test.runIf(PyRepo.supported)("workflow", { timeout: 30000 }, async () => {
-  using tmpDir = makeTmpDir();
-  await using repo = await PyRepo.create("/myrepo", { dir: tmpDir.name });
-
-  const store = new PyRepoStore({
+  store = new PyRepoStore({
     repoPrefix: new Name("/myrepo"),
     combineRange: true,
   });
-  const cOpts: ConsumerOptions = {
-    modifyInterest: { lifetime: 500 },
-    retx: 1,
-  };
+});
+afterEach(() => {
+  closers.close();
+  Forwarder.deleteDefault();
+});
+
+const cOpts: ConsumerOptions = {
+  modifyInterest: { lifetime: 500 },
+  retx: 1,
+};
+
+test.runIf(PyRepo.supported)("basic", { timeout: 30000, retry: 2 }, async () => {
+  const readable = new ReadFromNetwork(cOpts).mix(store);
+  await testDataStoreBasic(readable);
+});
 
+test.runIf(PyRepo.supported)("workflow", { timeout: 30000, retry: 1 }, async () => {
   const names = Array.from({ length: 200 }, (item, i) => {
     void item;
     if (i < 100) {
@@ -35,20 +53,20 @@ test.runIf(PyRepo.supported)("workflow", { timeout: 30000 }, async () => {
     }
     return new Name(`/Z/${i}`);
   });
-  const pkts = await Promise.all(Array.from(names, async (name) => {
-    const data = new Data(name);
-    await digestSigning.sign(data);
-    return data;
-  }));
 
   const countRetrievable = async () => {
     const retrieved = await Promise.allSettled(Array.from(names, (name) => consume(name, cOpts)));
     return retrieved.filter(({ status }) => status === "fulfilled").length;
   };
 
-  await store.insert(pkts);
+  await store.insert(parallelMap(Infinity, async (name) => {
+    const data = new Data(name);
+    await digestSigning.sign(data);
+    return data;
+  }, names));
   await expect(countRetrievable()).resolves.toBeGreaterThanOrEqual(names.length * 0.8);
 
+  await delay(1000);
   await store.delete(...names);
   await expect(countRetrievable()).resolves.toBe(0);
 });

pkg/repo-api/package.json

@@ -26,6 +26,7 @@
     "types": "lib/mod.d.ts"
   },
   "dependencies": {
+    "@ndn/endpoint": "workspace:*",
     "@ndn/l3face": "workspace:*",
     "@ndn/naming-convention2": "workspace:*",
     "@ndn/packet": "workspace:*",

pkg/repo-api/src/mod.ts

@@ -3,6 +3,7 @@ export * from "./bulk-insert-target";
 export * from "./copy";
 export * from "./data-array";
 export * from "./data-tape";
+export * from "./read-from-network";
 export * from "./respond-rdr";
 
 /**

pkg/repo-api/src/read-from-network.ts

@@ -0,0 +1,42 @@
+import { consume, type ConsumerOptions } from "@ndn/endpoint";
+import { type Data, Interest, type Name } from "@ndn/packet";
+
+import type * as S from "./data-store";
+
+/** Implement DataStore read-side interfaces by sending Interests to the network. */
+export class ReadFromNetwork implements S.Get, S.Find {
+  constructor(private readonly cOpts?: ConsumerOptions) {}
+
+  public get(name: Name): Promise<Data | undefined> {
+    return this.find(new Interest(name));
+  }
+
+  public async find(interest: Interest): Promise<Data | undefined> {
+    try {
+      return await consume(interest, this.cOpts);
+    } catch {
+      return undefined;
+    }
+  }
+
+  /**
+   * Extend a write-only DataStore with read methods.
+   * @param inner - Inner DataStore that does not support reading.
+   * @returns Readable DataStore.
+   */
+  public mix<T extends {}>(inner: T): T & S.Get & S.Find {
+    const self = this; // eslint-disable-line unicorn/no-this-assignment, @typescript-eslint/no-this-alias
+    return new Proxy<any>(inner, {
+      get(target, prop) {
+        void target;
+        switch (prop) {
+          case "get":
+          case "find": {
+            return (self as any)[prop];
+          }
+        }
+        return (inner as any)[prop];
+      },
+    });
+  }
+}

pkg/repo-api/test-fixture/data-store.ts

@@ -5,36 +5,41 @@ import { Data, Interest, Name, type NameLike } from "@ndn/packet";
 import { BufferChunkSource, fetch, serve } from "@ndn/segmented-object";
 import { makeObjectBody } from "@ndn/segmented-object/test-fixture/object-body";
 import { collect, map } from "streaming-iterables";
+import type { RequireAtLeastOne } from "type-fest";
 import { expect } from "vitest";
 
 import type { DataStore as S } from "..";
 
-export async function testDataStoreBasic(store: S.Insert & Partial<S.ListNames & S.ListData & S.Get & S.Find & S.Delete>): Promise<void> {
+type SReadable = S.ListNames & S.ListData & S.Get & S.Find;
+
+export async function testDataStoreBasic(store: RequireAtLeastOne<S.Insert & Partial<SReadable & S.Delete>, keyof SReadable>): Promise<void> {
   const checkNames = async (prefix: NameLike | undefined, positive: Iterable<NameLike>, negative: Iterable<NameLike> = []) => {
     prefix = prefix ? Name.from(prefix) : undefined;
-    if (store.listNames) {
-      await expect(collect(store.listNames(prefix))).resolves.toEqualNames(positive);
-    }
-    if (store.listData) {
-      await expect(collect(map((data) => data.name, store.listData(prefix))))
-        .resolves.toEqualNames(positive);
-    }
-    if (store.get) {
-      for (const name of positive) {
-        await expect(store.get(Name.from(name))).resolves.toHaveName(name);
+    await Promise.all((function*(): Iterable<Promise<void>> {
+      if (store.listNames) {
+        yield expect(collect(store.listNames(prefix))).resolves.toEqualNames(positive);
       }
-      for (const name of negative) {
-        await expect(store.get(Name.from(name))).resolves.toBeUndefined();
+      if (store.listData) {
+        yield expect(collect(map((data) => data.name, store.listData(prefix))))
+          .resolves.toEqualNames(positive);
       }
-    }
-    if (store.find) {
-      for (const name of positive) {
-        await expect(store.find(new Interest(name))).resolves.toHaveName(name);
+      if (store.get) {
+        for (const name of positive) {
+          yield expect(store.get(Name.from(name))).resolves.toHaveName(name);
+        }
+        for (const name of negative) {
+          yield expect(store.get(Name.from(name))).resolves.toBeUndefined();
+        }
       }
-      for (const name of negative) {
-        await expect(store.find(new Interest(name))).resolves.toBeUndefined();
+      if (store.find) {
+        for (const name of positive) {
+          yield expect(store.find(new Interest(name))).resolves.toHaveName(name);
+        }
+        for (const name of negative) {
+          yield expect(store.find(new Interest(name))).resolves.toBeUndefined();
+        }
       }
-    }
+    })());
   };
 
   await store.insert(new Data("/A/1"), new Data("/A/2"));