/ 
DBIO-0.900004
/ DBIO::Test::AsyncParity

NAME

DBIO::Test::AsyncParity - The same ResultSet operations, synchronous and asynchronous, on one database, compared

VERSION

version 0.900004

SYNOPSIS

package DBIO::PostgreSQL::EV::Test::AsyncParity;
use base 'DBIO::Test::AsyncParity';

sub parity_deviations {{
  'upsert' => {
    kind   => 'pending',                  # not built yet
    card   => 'k58',                      # required for pending
    reason => qr/upsert/,
  },
  'row-copy' => {
    kind   => 'unsupported',              # this transport cannot do it
    reason => qr/copy_async/,
  },
}}

# the seams with a working default, only when the dist needs its own:
sub parity_reset {
  my ($class, $schema, $harness) = @_;
  $class->SUPER::parity_reset($schema, $harness);
  $schema->storage->dbh_do(sub { $_[1]->do('ALTER SEQUENCE ... RESTART') });
}

1;

and in the test file DBIO::PostgreSQL::EV::Test::AsyncParity->run(...).

DESCRIPTION

An async transport (future_io, ev, ...) answers the same ORM surface a blocking connection does. This module is the one list of cases that proves it, stated once in core: every case is driven twice on the same connected schema and the same database -- once through the synchronous method, once through its *_async twin -- and what the two ways produced is compared. Compared are the rows (as plain data), the counts, the state the case left behind, and, for the cases that are about failing, the exception class.

The suite takes the harness (DBIO::Test::AsyncHarness) a transport already has and is called by a thin test file:

use DBIO::PostgreSQL::EV::TestHarness;
use DBIO::Test::AsyncParity;

DBIO::PostgreSQL::EV::TestHarness->skip_all_unless_live;
my $h = DBIO::PostgreSQL::EV::TestHarness->new(
  schema_class => 'DBIO::Test::Schema',
  storage_type => 'DBIO::PostgreSQL::Storage',  # the driver; the schema has none
);

DBIO::Test::AsyncParity->run(harness => $h, schema_class => 'DBIO::Test::Schema');
done_testing;

There are exactly three acceptable answers for a case:

pass

The two ways agree, and the case's own expectation (if it has one) holds.

deviation

The transport states, in "parity_deviations", that this case does not agree and why. A declared deviation is a test, not a skip: the case is driven, it has to fail the way the declaration says, and a declared deviation of a case that passes is a failure -- the declaration is stale.

absent

Core has no asynchronous twin for the operation at all (populate). The table says so itself; the case is driven, the missing twin is observed, and it turns red when core grows one.

Anything else is a failure, and a failure names the number and the name of the case. A case is not skipped by choice: the only exception is a missing capability ("Capabilities"), and then it is a test of its own, labelled skipped with the capability named, and listed on the closing line. Without a live harness the file is skipped as a whole by the harness, which is where that decision belongs.

The model is DBIO::Test::Conformance (ADR 0060): one table in core, the transport answers it, a deviation is declared and checked.

METHODS

table_version

The version of the case table, 2 (version 1 had the 41 cases of the ResultSet surface; 2 adds the cases of the async scaffold, 42 to 56).

cases

my $cases = DBIO::Test::AsyncParity->cases;

The arrayref of cases, in number order, a fresh copy. See DBIO::Test::AsyncParity::Cases.

parity_deviations

Seam with a default: none. See "Declaring deviations".

parity_seed

Seam with a working default, see "WRITING A TRANSPORT'S CLASS".

parity_reset

Seam with a working default, see "WRITING A TRANSPORT'S CLASS".

parity_capabilities

my @names = $class->parity_capabilities($harness);

The capabilities of the transport under test, the names a case lists in requires. The default gives what can be read off the transport: its transport_capabilities (listen, notify, ...), pipeline when it overrides the pipeline seams ("_pipeline_refusal" in DBIO::Storage::Async), cancel when its Future class can cancel, and pool_queue when its pool is a DBIO::Storage::PoolBase. A transport adds what only it knows (see "Capabilities") and calls this method first.

parity_fault

my $guard = $class->parity_fault($schema, $harness, $name, @args);

Make the transport fail in the way $name says, until the returned guard goes out of scope. The default throws: a transport that offers a fault lists fault:NAME in "parity_capabilities" and implements it here. Names: rollback_fails (a ROLLBACK of a transaction is not confirmed), disconnect (the connection of the transaction context passed as the first argument is lost).

parity_session_sql

my ($set, $probe) = $class->parity_session_sql($harness);

Two statements for the case session-reset: one that leaves state in the session of the connection it runs on, and one that fails when that state is gone. The default is a temporary table, CREATE TEMPORARY TABLE and a SELECT from it. An engine without that syntax overrides it.

parity_unreachable

my $schema = $class->parity_unreachable($harness, $password);

A connected schema in the harness's async mode whose connection cannot be made, built with $password; the case connect-error-masked asks it for the count of artists and checks the error does not carry the password. The default throws; a transport that implements it lists unreachable_connection in "parity_capabilities".

parity_sync_schema

Seam with a working default, see "WRITING A TRANSPORT'S CLASS".

check_declarations

my @problems = $class->check_declarations;

What can be checked without a server: every declared deviation names a case of the table, has kind unsupported, pending or absent, a regex reason, a card for pending, and does not redeclare a case the table itself marks absent. Returns the problems as strings, none when the declaration is whole. "run" makes it the first test.

results

my @results = $class->results(harness => $h, schema_class => 'DBIO::Test::Schema');
my @results = $class->results(harness => $h, only => [ 'find-pk' ]);

Drive every case against the harness's schema and return one hashref per case: { no, case, kind, status, message } and, for deviation, the deviation kind and for pending the card. It does not touch Test::More. status is pass, deviation, absent or fail. only restricts the run to the named cases.

schema_class names the schema the table was written for (default DBIO::Test::Schema); the runner only uses it to say what is missing when the connected schema lacks a source the table needs.

summary_line

my $line = $class->summary_line(\@results);

The one line that closes a run: the declared deviations with their cards and the absent operations. undef when there is nothing to say.

run

DBIO::Test::AsyncParity->run(harness => $h, schema_class => 'DBIO::Test::Schema');

Write the tests: first one for the declarations ("check_declarations"), then one per case, labelled with the case number, its name and its status. A failure's message is printed with diag, the closing summary line too. Does not call done_testing.

WRITING A TRANSPORT'S CLASS

Every method is a class method. Nothing must be provided: a transport that answers every case needs no class of its own and calls DBIO::Test::AsyncParity->run directly.

parity_deviations

Hashref keyed by case name (see "CASES"), default {}. See "Declaring deviations".

parity_seed($schema, $harness)

Put the fixture of "fixture" in DBIO::Test::AsyncParity::Cases into the empty database: three artists, four CDs, four tracks. The default creates them through the ResultSet API, without naming a key, and works on any schema built from DBIO::Test::Schema. Override when the database wants something else (rows by COPY, a different schema).

parity_reset($schema, $harness)

Remove everything a case could have left and everything the seed made, so the next run starts empty. The default deletes the rows of OneKey, Track, CD and Artist, children first, with one DELETE each. It runs before each way of each case and once more after the case, also after a failure. Override it when the tables carry something it must not delete or when a sequence is to be put back.

parity_sync_schema($harness)

A second, purely synchronous connection to the same database, for the case that is about an *_async call on a connection without an async mode. The default connects the harness's schema class with the harness's own DSN, user and password and no async attribute. When the harness carries a storage_type (see "new" in DBIO::Test::AsyncHarness), it connects a clone of the schema class declaring that driver, as the harness's own schema does; otherwise the schema class must declare its driver itself (ADR 0067). Disconnected when the run ends.

Declaring deviations

parity_deviations returns a hashref keyed by case name:

{ case_name => { kind => 'unsupported' | 'pending' | 'absent',
                 reason => qr/.../, card => 'k58' } }
kind unsupported

The transport cannot do it, by design. Permanent; needs no card.

kind pending

The transport does not do it yet. card is required: a card of the transport's own board, in the form k58.

kind absent

The operation has no asynchronous twin. Cases of the table that are absent in core carry that themselves; a transport only uses it for its own gap.

reason

A regex the failure's message has to match. The message is what the runner reports for the case: the exception the asynchronous way threw, or the difference between the two ways.

The run ends with one line, printed by "run" and returned by "summary_line":

2 deviations: 34 upsert (pending, karr k58), 27 row-copy (unsupported); 1 absent: 41 populate

Capabilities

A case may list requires => [ 'name', ... ]. A transport that does not have every one of them does not run the case: its result is skipped, a passing test labelled NN name: skipped (needs NAME), and the closing line lists it. The capabilities of a transport are the names "parity_capabilities" answers; the default reads them off the transport:

listen, notify, ...  the transport's transport_capabilities
pipeline             it implements the pipeline seams
cancel               its Future class can cancel
pool_queue           its pool is a DBIO::Storage::PoolBase

A transport adds the ones only it knows, and implements the seams behind them:

fault:rollback_fails       parity_fault 'rollback_fails'
fault:disconnect           parity_fault 'disconnect'
unreachable_connection     parity_unreachable

A skipped case is a gap of that transport, not of the table. The capability is the single place that decides, so a case never skips by a driver name.

CASES

The table is DBIO::Test::AsyncParity::Cases. A case has a number, the name a declaration is keyed by, and a kind: direct (the synchronous method has an *_async twin), composed (core has none, the case composes the operation from the ones it has) or async-only (nothing to compare against; the case states what the asynchronous way must answer).

Reads
 1 find-pk             find by primary key
 2 find-unique         find through a named unique constraint (ADR 0034)
 3 find-miss           find that matches nothing: undef, no error
 4 find-rel-key        find whose key is a relationship value
 5 search-all          search with a condition, all rows
 6 order-rows-offset   order_by with rows and offset
 7 group-by            group_by over a column subset
 8 prefetch            has_many prefetch, collapsed
 9 join                join on a relationship
10 search-related      related resultset of a row (composed)
11 first               first row of an ordered set
12 single              single row by a condition
13 count               count of a condition
14 count-literal       count with a literal SQL condition
15 results-exist       the EXISTS question, hit and miss
16 rsc-all             all values of a column
17 rsc-single          single value of a column
18 rsc-min-max-sum     min, max and sum of a column
19 rsc-func            aggregate by name
Writes
20 create              create a row
21 create-cascade      create with nested related rows (composed)
22 row-insert          insert a new_result row
23 row-update          update a fetched row
24 row-update-rs-loop  update of a set, row by row (composed)
25 row-delete          delete a fetched row (OneKey)
26 row-delete-rs-loop  delete of a set, row by row (composed)
27 row-copy            copy a row (OneKey)
28 row-reload          get_from_storage and discard_changes
29 update-or-insert    update_or_insert of an unstored row
30 find-or-create      find_or_create of a missing row
31 find-or-new         find_or_new of a missing row
32 update-or-create    update_or_create of a present row
33 update-or-new       update_or_new of a present row
34 upsert              single-statement upsert (capability)
Transactions and context
35 txn-commit          txn_do / txn_do_async that commits
36 txn-rollback        txn_do / txn_do_async that fails and rolls back
37 txn-with-context    a ResultSet bound to the transaction context (composed)
38 txn-adopts-future   the returned Future is adopted (async-only)
Errors
39 err-unique          a unique violation: the same exception class
40 async-on-sync-connection   *_async without an async mode croaks (async-only)
No async twin
41 populate            named, not invented (absent)
Pipeline and cancel
42 pipeline-commit     a pipeline of two inserts (pipeline)
43 pipeline-fail       a failing statement fails the batch, the pool serves (pipeline)
44 cancel-txn          a cancelled transaction rolls back (cancel)
45 cancel-statement    a cancelled statement gives its connection back (cancel)
46 cancel-batch        a cancelled pipeline retires its connection (pipeline, cancel)
Failure inside a transaction
47 rollback-failed     a ROLLBACK not confirmed retires the connection (fault:rollback_fails)
48 txn-statement-in-flight        COMMIT waits for a statement still running
49 txn-statement-in-flight-fails  a dropped statement that fails fails the transaction
50 disconnect-in-txn   the connection is lost mid-transaction (fault:disconnect)
Pool and session
51 pool-self-deadlock  a call queued for the transaction's own pool is refused (pool_queue)
52 pool-max-waiters    the queue is bounded (pool_queue)
53 pool-acquire-timeout  a wait is bounded (pool_queue)
54 session-reset       session state is gone after release (pool_queue)
Transport
55 listen-notify       LISTEN and NOTIFY (listen, notify)
56 connect-error-masked  a connect error carries no password (unreachable_connection)

Cases 1 to 40 cover the ResultSet surface that has an async twin; 41 names the one operation of that surface that has no twin, so that its absence is a line of the run and not a gap in the list. Cases 42 to 56 run the pipeline, cancel, transaction, pool and transport behaviour of the async layer on a real transport. Not in the table: a disconnect in the middle of a deploy and the rotation of an AccessBroker's credentials need a harness of their own; core covers both offline (t/test/64_async_deploy_across_disconnect.t, t/test/63_async_pool_set_access_broker_built.t).

AUTHOR

DBIO & DBIx::Class Authors

COPYRIGHT AND LICENSE

Copyright (C) 2026 DBIO Authors Portions Copyright (C) 2005-2025 DBIx::Class Authors Based on DBIx::Class, heavily modified.

This is free software; you can redistribute it and/or modify it under the same terms as the Perl 5 programming language system itself.