Catcrafts/Crafter.Review
Watch
2
0
Fork
You've already forked Crafter.Review
 0
Code Issues 7 Pull requests Projects Releases 3 Packages Wiki Activity Actions

test(run): the store rules, stacks and memory, and the review fixes, check by check

Browse source 
StoreRules (23 checks), StacksMemory (24) and ReviewFixes (19), each a
test of its own.

Run and SlotsPlan now have seams that start as the real ones: appReady,
probe, query, bypassesRls, migrationsAtHead, migrationsDrift, treeDrift,
promptFor, runPass, resumable, instrumentPreflight, and a Machine for
SlotsPlan. MigrationsAtHead is split out of MigrationsDrift for it; the
HEAD side is still worked out once per tree and base.

The Postgres stand-in and the raw broker call move to tests/Common, shared
by PgWire and ReviewFixes. ReviewFixes runs this binary as `sql`, and a
real server that resets every connection makes the startup message either
fail to go out or never get its answer - a DbError either way.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Jorijn van der Graaf 2026-10-31 09:39:53 +01:00
commit dd0929b705
Signed by: jorijnvdgraaf
GPG key ID: 2937E59CDCC1BCFB
8 changed files with 1328 additions and 185 deletions
Download patch file Download diff file Expand all files Collapse all files

150
tests/PgWire/main.cpp
View file

@ -2,8 +2,8 @@
// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®
// The SQL helper: its statements, refusals, the URL, SCRAM, the protocol on a
// stand-in server (FakePg, below). Then what the helper prints over the same
// stand-in, line for line against the goldens (fixture/pipeline.json,
// stand-in server (FakePg, tests/Common). Then what the helper prints over
// the same stand-in, line for line against the goldens (fixture/pipeline.json,
// `runs`), and the broker: the URL stays in the driver, a pass gets an
// address and a token.
@ -16,6 +16,9 @@ import Crafter.Review;
namespace fs = std::filesystem;
using namespace Crafter::Review;
#include "../Common/FakePg.hpp"
using namespace Testing;
namespace {
std::int32_t Failures = 0;
@ -24,134 +27,6 @@ namespace {
if (!cond) ++Failures;
}
std::string Be32(std::uint32_t v) {
return {static_cast<char>(v >> 24), static_cast<char>((v >> 16) & 0xFF), static_cast<char>((v >> 8) & 0xFF), static_cast<char>(v & 0xFF)};
}
std::string Msg(char type, const std::string& body = {}) {
return std::format("{}{}{}", type, Be32(static_cast<std::uint32_t>(body.size() + 4)), body);
}
std::string Err(const std::string& code, const std::string& text) {
return Msg('E', std::format("SERROR{}C{}{}M{}{}{}", std::string(1, '\0'), code, std::string(1, '\0'), text, std::string(1, '\0'), std::string(1, '\0')));
}
// A Postgres stand-in on 127.0.0.1 for one connection: cleartext
// password, simple Query, Parse/Bind/Describe/Execute/Sync. It records
// what it is sent and answers as the server does where the helper relies
// on it: two commands in one Parse are an error; COMMIT ends the
// transaction (status I); a statement after an error fails.
class FakePg {
public:
explicit FakePg(std::string password = "pw") : password(std::move(password)) {
listener = socket(AF_INET, SOCK_STREAM, 0);
sockaddr_in address {};
address.sin_family = AF_INET;
address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
bind(listener, reinterpret_cast<sockaddr*>(&address), sizeof(address));
listen(listener, 1);
socklen_t length = sizeof(address);
getsockname(listener, reinterpret_cast<sockaddr*>(&address), &length);
port = ntohs(address.sin_port);
thread = std::thread([this] { Serve(); });
}
~FakePg() {
Served();
}
// The log once the connection is over.
std::vector<std::pair<std::string, std::string>> Served() {
if (thread.joinable()) thread.join();
return log;
}
std::uint16_t port = 0;
private:
std::string password;
std::int32_t listener = -1;
std::thread thread;
std::vector<std::pair<std::string, std::string>> log;
void Serve() {
const std::int32_t c = accept(listener, nullptr, nullptr);
std::string buffer;
auto read = [&](std::size_t n) {
while (buffer.size() < n) {
std::array<char, 4096> chunk {};
const ssize_t got = recv(c, chunk.data(), chunk.size(), 0);
if (got <= 0) break;
buffer.append(chunk.data(), static_cast<std::size_t>(got));
}
const std::string out = buffer.substr(0, std::min(n, buffer.size()));
buffer.erase(0, out.size());
return out;
};
auto length = [](const std::string& bytes) {
return (static_cast<std::uint32_t>(static_cast<std::uint8_t>(bytes[0])) << 24) | (static_cast<std::uint32_t>(static_cast<std::uint8_t>(bytes[1])) << 16)
| (static_cast<std::uint32_t>(static_cast<std::uint8_t>(bytes[2])) << 8) | static_cast<std::uint8_t>(bytes[3]);
};
auto send = [&](const std::string& data) { ::send(c, data.data(), data.size(), MSG_NOSIGNAL); };
const std::uint32_t startup = length(read(4));
read(startup - 4);
send(Msg('R', Be32(3)));
read(1);
std::string got = read(length(read(4)) - 4);
while (!got.empty() && got.back() == '\0') got.pop_back();
log.emplace_back("password", got);
if (got != password) {
send(Err("28P01", "password authentication failed"));
close(c);
close(listener);
return;
}
send(Msg('S', std::string("server_version\x00" "17.6\x00", 20)) + Msg('R', Be32(0)) + Msg('Z', "I"));
std::string status = "I";
std::string pending;
while (true) {
const std::string type = read(1);
if (type.empty()) break;
const std::string body = read(length(read(4)) - 4);
if (type == "Q") {
std::string q = body;
while (!q.empty() && q.back() == '\0') q.pop_back();
log.emplace_back("Q", q);
status = q.starts_with("BEGIN") ? "T" : (q == "ROLLBACK" ? "I" : status);
send(Msg('C', Dyn::Split(q)[0] + std::string(1, '\0')) + Msg('Z', status));
} else if (type == "P") {
pending = body.substr(1, body.find('\0', 1) - 1);
log.emplace_back("P", pending);
} else if (type == "S") {
const std::string low = Dyn::Lower(pending);
std::string out;
if (status == "E") {
out = Err("25P02", "current transaction is aborted");
} else if (pending.contains(';')) {
out = Err("42601", "cannot insert multiple commands into a prepared statement");
status = "E";
} else if (low.starts_with("commit")) {
out = Msg('C', std::string("COMMIT\0", 7));
status = "I";
} else if (low.starts_with("select")) {
out = Msg('T', std::string("\x00\x01x\x00", 4) + std::string(18, '\0')) + Msg('D', std::string("\x00\x01\x00\x00\x00\x01" "1", 7))
+ Msg('N', std::string("SNOTICE\0Mhello\0\0", 16)) + Msg('C', std::string("SELECT 1\0", 9));
} else if (low.starts_with("boom")) {
out = Err("42601", "syntax error at or near \"boom\"");
status = "E";
} else {
out = Msg('C', Dyn::Upper(Dyn::Split(pending)[0]) + std::string(1, '\0'));
}
send(Msg('1') + Msg('2') + out + Msg('Z', status));
} else if (type == "X") {
log.emplace_back("X", "");
break;
}
}
close(c);
close(listener);
}
};
std::uint16_t DeadPort() {
const std::int32_t s = socket(AF_INET, SOCK_STREAM, 0);
@ -185,20 +60,7 @@ namespace {
// A raw call of the broker: one line in, every line back.
std::vector<std::string> RawCall(std::uint16_t port, const std::string& line) {
const std::int32_t s = socket(AF_INET, SOCK_STREAM, 0);
sockaddr_in address {};
address.sin_family = AF_INET;
address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
address.sin_port = htons(port);
connect(s, reinterpret_cast<sockaddr*>(&address), sizeof(address));
::send(s, line.data(), line.size(), MSG_NOSIGNAL);
shutdown(s, SHUT_WR);
std::string all;
std::array<char, 4096> chunk {};
ssize_t got = 0;
while ((got = recv(s, chunk.data(), chunk.size(), 0)) > 0) all.append(chunk.data(), static_cast<std::size_t>(got));
close(s);
return Dyn::Split(Dyn::RStrip(all, "\n"), "\n");
return Dyn::Split(Dyn::RStrip(BrokerCall(port, line), "\n"), "\n");
}
}