From d6f9d135b7ddb9f4ab59fabc001238b28df299e3 Mon Sep 17 00:00:00 2001 From: Can Tuncay Date: Thu, 16 Jul 2026 14:50:31 -0500 Subject: [PATCH] Revert "Added per-destination event filtering (#56)" This reverts commit b1c17425167025283d4bcbdc640c50c485122f57. --- CLAUDE.md | 106 -------------------------------- README.md | 11 ---- emulator.cabal | 1 - src/Ambar/Emulator.hs | 1 - src/Ambar/Emulator/Config.hs | 22 ------- src/Ambar/Emulator/Projector.hs | 75 +++------------------- tests/Test/Config.hs | 78 ----------------------- tests/Test/Emulator.hs | 60 +----------------- tests/Test/Projector.hs | 52 ---------------- tests/Tests.hs | 2 - 10 files changed, 10 insertions(+), 398 deletions(-) delete mode 100644 CLAUDE.md delete mode 100644 tests/Test/Projector.hs diff --git a/CLAUDE.md b/CLAUDE.md deleted file mode 100644 index ae4cb80..0000000 --- a/CLAUDE.md +++ /dev/null @@ -1,106 +0,0 @@ -# CLAUDE.md - -This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. - -## Overview - -The Ambar event-bus **emulator**: a local stand-in for the hosted Ambar platform. It polls rows out of source databases (PostgreSQL / MySQL / SQL Server / plain files), stores them in a file-backed partitioned queue, and POSTs each row to registered HTTP destinations with per-destination cursors and infinite retry. Consuming apps (e.g. `sumori`, `virtual-agent`) run it via docker-compose as their local event bus. - -Written in **Haskell** (GHC2021). No relation to the TypeScript app repos' conventions — this repo follows ordinary Haskell/cabal practice. - -## Dev environment (macOS, arm64) - -```bash -# Toolchain — match CI: GHC 9.10.1 (cabal 3.10+ works) -ghcup install ghc 9.10.1 && ghcup set ghc 9.10.1 - -# The vendored libs are a git submodule — nothing builds without this: -git submodule update --init - -# Native libraries the Haskell bindings link against: -brew install libpq mysql-client openssl@3 zstd pcre -``` - -Homebrew's libpq/openssl are keg-only, and cabal's `--extra-*-dirs` CLI flags do **not** propagate to dependency configure steps — put the paths in the user-global cabal config (`~/.config/cabal/config`) instead: - -``` -extra-include-dirs: /opt/homebrew/opt//include -extra-lib-dirs: /opt/homebrew/opt//lib -``` - -for each of `libpq`, `mysql-client`, `openssl@3`, `zstd`, `pcre`. Also keep `/opt/homebrew/opt/libpq/bin` and `/opt/homebrew/opt/mysql-client/bin` on `PATH` (for `pg_config` / `mysql_config`). - -## Commands - -```bash -./utils.sh build # cabal build lib:emulator -./utils.sh run # run the emulator (needs --config FILE) -./utils.sh test # cabal run emulator-tests (see test infra below) -./utils.sh test -- --match "some description" # subset by hspec description -./utils.sh typecheck # ghcid fast-feedback loop on the library -./utils.sh bench # benchmarks -./utils.sh build-docker # release image (static Alpine binary, build/Dockerfile.static) -``` - -Tests expect live **PostgreSQL and MySQL** servers (CI provisions Postgres 16 + MySQL 9; see `.github/workflows/build.yml` for the exact setup). Config/queue/projector suites are pure and run without databases — use `--match` to scope. SQL Server needs no provisioning: the harness `docker run`s `mcr.microsoft.com/azure-sql-edge` itself (works on Apple Silicon). - -### Running the full suite locally on macOS (Docker-backed DBs) - -The harnesses shell out to `psql`/`mysql` and self-provision users/databases, with three macOS traps: the Haskell `mysql` library always dials the Unix socket `/tmp/mysql.sock` (host `localhost`), Unix sockets can't cross the macOS↔Docker-VM boundary, and MySQL 9's `caching_sha2_password` rejects cold-cache auth over a bridged socket (client treats a socket as a secure channel and sends cleartext; the server sees insecure TCP and refuses it). Verified recipe (286/286 pass, 2026-07-16): - -```bash -# Throwaway servers — port 5432 must be free (stop any app stack using it first) -docker run -d --rm --name emu-test-pg -e POSTGRES_HOST_AUTH_METHOD=trust -p 5432:5432 postgres:16 -# mysql:8.0 + native password sidesteps the caching_sha2 cold-cache failure (8.0-only flag; removed in 9) -docker run -d --rm --name emu-test-mysql -e MYSQL_ALLOW_EMPTY_PASSWORD=yes -p 3306:3306 \ - mysql:8.0 --default-authentication-plugin=mysql_native_password - -# Bridge the socket the Haskell client insists on to the container's TCP port. -# backlog=511 matters: hspec runs specs in parallel and socat's default backlog of 5 -# overflows into ECONNREFUSED mid-suite. -socat UNIX-LISTEN:/tmp/mysql.sock,fork,reuseaddr,backlog=511 TCP:127.0.0.1:3306 & - -PATH="$HOME/.ghcup/bin:/opt/homebrew/opt/libpq/bin:/opt/homebrew/opt/mysql-client/bin:$PATH" \ - PGHOST=127.0.0.1 PGUSER=postgres cabal test emulator-tests --test-show-details=direct -``` - -## Architecture - -``` -source DB ──(polling SELECT, RepeatableRead)──▶ topic (file-backed, partitioned) - │ one topic per source, partitioned by partitioningColumn - ▼ - one Projector per destination (consumer group = destination id) - │ per-partition consumers, per-destination cursor - │ optional per-destination filter (skip-and-commit) - ▼ - HTTP POST Message{data_source_id, …, payload: } - (infinite fibonacci-backoff retry, commit after success) -``` - -| Concern | Where | -|---|---| -| Orchestration (`emulate`) | `src/Ambar/Emulator.hs` | -| Config schema + YAML parsing | `src/Ambar/Emulator/Config.hs` | -| Delivery loop, filtering, envelope | `src/Ambar/Emulator/Projector.hs` | -| Polling connectors | `src/Ambar/Emulator/Connector/{Postgres,MySQL,MicrosoftSQLServer,Poll}.hs` | -| File-backed queue/topics | `src/Ambar/Emulator/Queue/…` | -| HTTP transport (auth, retry policy decode) | `src/Ambar/Transport/Http.hs` | -| Source cursors (`state.json`, saved every 30s) | `src/Ambar/Emulator.hs` | - -Key semantics: - -- **At-least-once, per-partition ordered** delivery; a destination's consumer commits only after a successful send (or an intentional filter skip). -- **Per-destination filter** (optional `filter: {column, values}` in config): non-matching records are skipped but committed. Records missing the column, or with a non-string value, are **delivered** (fail-open) — config mistakes must degrade to the pre-filter behaviour, never to silent data loss. -- The full source row (all configured `columns`) is the `payload`; consumers see it wrapped in the `Message` envelope. - -## Consumers of this repo - -- Apps reference the released image `docker.io/ambarltd/emulator:vX.Y` in their docker-compose (`event-bus` service) and mount their config at `/opt/emulator/config/config.yaml`. -- Releases: git tags (`v1.x`) + static Docker image via `./utils.sh build-docker`; CI (`.github/workflows/`) builds and tests on push. -- The consuming apps' config registers one destination per projection/reaction endpoint. Keep any destination `filter.values` lists in lockstep with what the consumer actually accepts — a stale filter silently starves the consumer (the apps should own a drift check on their side). - -## Git - -- Feature branches; never commit to `main` or push without explicit approval. -- `deps/haskell-libs` is a pinned submodule — don't update its ref as a side effect. diff --git a/README.md b/README.md index 74375a1..692784c 100644 --- a/README.md +++ b/README.md @@ -102,17 +102,6 @@ data_destinations: - sqlserver_source - file_source - # Optional. Deliver only records whose `column` value is in `values`; - # other records are skipped (the destination's cursor still advances). - # Omit to deliver every record. Records missing the column (or with a - # non-string value in it) are always delivered — a misconfigured column - # name degrades to full delivery, never to silently dropped records. - filter: - column: event_name - values: - - OrderCreated - - OrderUpdated - # Send data to a file. One entry per line. - id: file_destination description: my projection 1 diff --git a/emulator.cabal b/emulator.cabal index 448794b..8396606 100644 --- a/emulator.cabal +++ b/emulator.cabal @@ -112,7 +112,6 @@ test-suite emulator-tests main-is: Tests.hs other-modules: Test.Config - Test.Projector Test.Queue Test.Connector Test.Connector.File diff --git a/src/Ambar/Emulator.hs b/src/Ambar/Emulator.hs index 27f19df..914bd50 100644 --- a/src/Ambar/Emulator.hs +++ b/src/Ambar/Emulator.hs @@ -123,7 +123,6 @@ emulate logger_ config env = do , p_destinationDescription = d_description dest , p_sources = sourceTopics , p_transport = transport - , p_filter = d_filter dest } withDestination dest act = diff --git a/src/Ambar/Emulator/Config.hs b/src/Ambar/Emulator/Config.hs index 8e879ac..c95cf23 100644 --- a/src/Ambar/Emulator/Config.hs +++ b/src/Ambar/Emulator/Config.hs @@ -8,7 +8,6 @@ module Ambar.Emulator.Config , Source(..) , DataDestination(..) , Destination(..) - , DestinationFilter(..) , Port(..) ) where @@ -27,8 +26,6 @@ import qualified Data.Text as Text import Data.Text (Text) import Data.Map.Strict (Map) import qualified Data.Map.Strict as Map -import Data.Set (Set) -import qualified Data.Set as Set import qualified Data.Yaml as Yaml import Ambar.Emulator.Connector.Poll (PollingInterval(..)) @@ -75,16 +72,6 @@ data DataDestination = DataDestination , d_sources :: [DataSource] , d_description :: Text , d_destination :: Destination - , d_filter :: Maybe DestinationFilter - } - --- | Server-side record filter for a destination. Only records whose --- @f_column@ value is one of @f_values@ are delivered; everything else is --- skipped (the consumer cursor still advances). A destination without a --- filter receives every record, as before. -data DestinationFilter = DestinationFilter - { f_column :: Text - , f_values :: Set Text } data Destination @@ -200,7 +187,6 @@ parseDataDestination sourcesMap = Json.withObject "DataSource" $ \o -> do [ "Invalid data destination type: '" <> t <> "'." , "Expected one of: http-push, file." ] - d_filter <- o .:? "filter" return DataDestination{..} where parseHTTPPush o = do @@ -211,14 +197,6 @@ parseDataDestination sourcesMap = Json.withObject "DataSource" $ \o -> do parseFile o = DestinationFile <$> (o .: "path") -instance FromJSON DestinationFilter where - parseJSON = Json.withObject "DestinationFilter" $ \o -> do - f_column <- o .: "column" - values <- o .: "values" - when (null values) $ - fail "filter.values must not be empty (omit the filter to deliver everything)" - return $ DestinationFilter f_column (Set.fromList values) - parseEnvConfigFile :: FilePath -> IO EnvironmentConfig parseEnvConfigFile path = do bs <- BS.readFile path diff --git a/src/Ambar/Emulator/Projector.hs b/src/Ambar/Emulator/Projector.hs index 35537a9..46c165a 100644 --- a/src/Ambar/Emulator/Projector.hs +++ b/src/Ambar/Emulator/Projector.hs @@ -1,8 +1,6 @@ module Ambar.Emulator.Projector ( Projection(..) , project - , filterVerdict - , FilterVerdict(..) , Message(..) , Payload(..) ) where @@ -10,10 +8,7 @@ module Ambar.Emulator.Projector {-| A projector reads messages from multiple queues, applies a filter to the stream and submits passing messages to a single data destination. -A destination with no filter receives every record. A destination with a -filter receives only records whose filter-column value is in the allowed -set; filtered-out records are committed without being sent, so the -consumer cursor always advances. +We do not implement filters for now. -} import qualified Data.Aeson as Json @@ -26,14 +21,10 @@ import Data.Text (Text) import qualified Data.Text.Encoding as Text import qualified Data.Text as Text import Control.Concurrent.Async (forConcurrently_) -import Control.Monad (when) import Control.Monad.Extra (whileM) -import Data.IORef (newIORef, atomicModifyIORef') import GHC.Generics (Generic) -import qualified Data.Set as Set - -import Ambar.Emulator.Config (Id(..), DataDestination, DataSource(..), Source(..), DestinationFilter(..)) +import Ambar.Emulator.Config (Id(..), DataDestination, DataSource(..), Source(..)) import Ambar.Emulator.Queue.Topic (Topic, ReadError(..), PartitionCount(..)) import qualified Ambar.Emulator.Queue.Topic as Topic import Ambar.Emulator.Connector.MicrosoftSQLServer (SQLServer(..)) @@ -53,7 +44,6 @@ data Projection = Projection , p_destinationDescription :: Text , p_sources :: [(DataSource, Topic)] , p_transport :: Some Transport - , p_filter :: Maybe DestinationFilter } -- | A record enriched with more information to send to the client. @@ -72,27 +62,14 @@ newtype Payload = Payload Json.Value deriving newtype (ToJSON, FromJSON) project :: SimpleLogger -> Projection -> IO () -project logger_ Projection{..} = do - -- One warning per destination per emulator run when the filter fails - -- open (reviewer note on PR #56): a typo'd filter column silently - -- restores full delivery, which looks exactly like a normal match. - -- Warning on every record would flood the log at full event rate, so - -- the first fail-open claims this flag and later ones stay quiet. - warnedFailOpen <- newIORef False - let warnFailOpenOnce logger reason = do - firstWarn <- atomicModifyIORef' warnedFailOpen (\claimed -> (True, not claimed)) - when firstWarn $ - logWarn logger $ - "filter fail-open: " <> reason - <> ". Delivering the record; the destination filter is not being applied." - <> " Further fail-open warnings for this destination are suppressed." - forConcurrently_ p_sources (projectSource warnFailOpenOnce) +project logger_ Projection{..} = + forConcurrently_ p_sources projectSource where - projectSource warnFailOpenOnce (source, topic) = + projectSource (source, topic) = -- one consumer per partition Topic.withConsumers topic group pcount $ \consumers -> forConcurrently_ consumers $ \consumer -> - whileM $ consume warnFailOpenOnce logger consumer source + whileM $ consume logger consumer source where PartitionCount pcount = Topic.partitionCount topic logger = @@ -100,7 +77,7 @@ project logger_ Projection{..} = do annotate ("dst: " <> unId p_destination) logger_ - consume warnFailOpenOnce logger consumer source = do + consume logger consumer source = do r <- Topic.read consumer case r of Left EndOfPartition -> return False @@ -108,15 +85,8 @@ project logger_ Projection{..} = do Right (bs, meta) -> do record <- decode logger bs let logger' = annotate (relevantFields (s_source source) record) logger - send = do - retrying logger' $ Transport.sendJSON p_transport (toMsg source record) - logInfo logger' ("sent." :: Text) - case filterVerdict p_filter record of - Matched -> send - Skipped -> logInfo logger' ("filtered." :: Text) - FailedOpen reason -> do - warnFailOpenOnce logger' reason - send + retrying logger' $ Transport.sendJSON p_transport (toMsg source record) + logInfo logger' ("sent." :: Text) Topic.commit consumer meta return True @@ -139,33 +109,6 @@ project logger_ Projection{..} = do fatal logger $ "decoding error: " <> err <> "\nraw: " <> raw Right v -> return v --- | The filter's decision for a record. --- --- 'FailedOpen' means the record is DELIVERED even though the filter could --- not be applied (missing column, null or non-string value, non-object --- record): a misconfigured column name must degrade to full delivery (the --- pre-filter behaviour), never to silently dropped records. The reason is --- carried so the projector can warn that the filter is not doing its job. -data FilterVerdict - = Matched -- ^ deliver: no filter, or the column value is allowed - | Skipped -- ^ commit without delivering - | FailedOpen Text -- ^ deliver, but the filter could not be applied - deriving (Eq, Show) - -filterVerdict :: Maybe DestinationFilter -> Payload -> FilterVerdict -filterVerdict Nothing _ = Matched -filterVerdict (Just DestinationFilter{..}) (Payload value) = - case value of - Json.Object o -> - case KeyMap.lookup (fromString $ Text.unpack f_column) o of - Just (Json.String v) - | Set.member v f_values -> Matched - | otherwise -> Skipped - Just Json.Null -> FailedOpen $ "filter column '" <> f_column <> "' is null" - Just _ -> FailedOpen $ "filter column '" <> f_column <> "' has a non-string value" - Nothing -> FailedOpen $ "filter column '" <> f_column <> "' is missing from the record" - _ -> FailedOpen "record is not a JSON object" - -- | Fields to print when a record is sent. relevantFields :: Source -> Payload -> Text relevantFields source (Payload value) = renderPretty $ diff --git a/tests/Test/Config.hs b/tests/Test/Config.hs index 557dd5e..caa6a87 100644 --- a/tests/Test/Config.hs +++ b/tests/Test/Config.hs @@ -4,7 +4,6 @@ module Test.Config (testConfig) where import Control.Exception (ErrorCall(..), fromException) import Data.List (isInfixOf) import qualified Data.Map.Strict as Map -import qualified Data.Set as Set import Data.String.Interpolate (i) import System.IO.Temp (withSystemTempFile) import System.IO (hClose) @@ -111,83 +110,6 @@ testConfig = do annotate "source count" $ Map.size (c_sources config) `shouldBe` 4 annotate "source count" $ Map.size (c_destinations config) `shouldBe` 2 - it "parses a destination filter" $ do - config <- parseConfig [i| - data_sources: - - id: file_source - description: The file source - type: file - path: ./source.txt - incrementingField: id - partitioningField: aggregate_id - - data_destinations: - - id: filtered_destination - description: filtered projection - type: file - path: ./temp.file - sources: - - file_source - filter: - column: event_name - values: - - OrderCreated - - OrderUpdated - |] - dest <- case Map.lookup (Id "filtered_destination") (c_destinations config) of - Nothing -> fail "destination missing" - Just d -> return d - case d_filter dest of - Nothing -> fail "expected a filter" - Just f -> do - annotate "filter column" $ f_column f `shouldBe` "event_name" - annotate "filter values" $ f_values f `shouldBe` Set.fromList ["OrderCreated", "OrderUpdated"] - - it "a destination without a filter parses to no filter" $ do - config <- parseConfig [i| - data_sources: - - id: file_source - description: The file source - type: file - path: ./source.txt - incrementingField: id - partitioningField: aggregate_id - - data_destinations: - - id: plain_destination - description: unfiltered projection - type: file - path: ./temp.file - sources: - - file_source - |] - dest <- case Map.lookup (Id "plain_destination") (c_destinations config) of - Nothing -> fail "destination missing" - Just d -> return d - annotate "no filter" $ null (d_filter dest) `shouldBe` True - - it "rejects an empty filter values list" $ do - parseConfig [i| - data_sources: - - id: file_source - description: The file source - type: file - path: ./source.txt - incrementingField: id - partitioningField: aggregate_id - - data_destinations: - - id: filtered_destination - description: filtered projection - type: file - path: ./temp.file - sources: - - file_source - filter: - column: event_name - values: [] - |] `shouldThrow` errorWith "must not be empty" - it "detects duplicate sources" $ do parseConfig [i| data_sources: diff --git a/tests/Test/Emulator.hs b/tests/Test/Emulator.hs index a898502..e1307dc 100644 --- a/tests/Test/Emulator.hs +++ b/tests/Test/Emulator.hs @@ -7,19 +7,15 @@ import Control.Exception (ErrorCall(..), throwIO) import Control.Concurrent.STM (newTVarIO, modifyTVar, atomically, retry, readTVar, writeTVar) import Control.Monad (forM, unless, void) import qualified Data.Aeson as Json -import qualified Data.Aeson.KeyMap as KeyMap import qualified Data.ByteString.Lazy as LB import qualified Data.Map.Strict as Map -import qualified Data.Set as Set -import Data.Text (Text) -import System.IO (hClose) import Test.Hspec ( Spec , it , describe , shouldBe ) -import System.IO.Temp (withSystemTempDirectory, withSystemTempFile) +import System.IO.Temp (withSystemTempDirectory) import Ambar.Emulator (emulate) import Ambar.Emulator.Config @@ -27,7 +23,6 @@ import Ambar.Emulator.Config , EmulatorConfig(..) , DataSource(..) , DataDestination(..) - , DestinationFilter(..) , Id(..) , Destination(..) , Source(..) @@ -58,52 +53,6 @@ testEmulator p = describe "emulator" $ do consumed <- consume out (length events) consumed `shouldBe` events - it "applies destination filters and continues past skipped records" $ - withConfig $ \config -> - withSystemTempFile "file-source-XXXXX" $ \path h -> do - hClose h - let source = DataSource - { s_id = Id "file_src" - , s_description = "file source" - , s_source = SourceFile - { sf_path = path - , sf_partitioningField = "aggregate_id" - , sf_incrementingField = "id" - } - } - -- same aggregate_id → one partition → a filtered record that - -- failed to commit would block every record behind it. - entry :: Int -> Text -> Json.Value - entry n name = Json.object - [ "id" Json..= n, "aggregate_id" Json..= (1 :: Int), "event_name" Json..= name ] - rows = [ entry 1 "Skipped", entry 2 "Wanted", entry 3 "Skipped", entry 4 "Wanted" ] - LB.writeFile path $ LB.intercalate "\n" (map Json.encode rows) <> "\n" - out <- newTVarIO [] - let dest = DataDestination - { d_id = Id "filtered_fun" - , d_sources = [source] - , d_description = "filtered destination" - , d_destination = DestinationFun $ \e -> do - atomically $ modifyTVar out (e:) - return Nothing - , d_filter = Just DestinationFilter - { f_column = "event_name" - , f_values = Set.fromList ["Wanted"] - } - } - env = mkEnv [source] [dest] - withAsyncThrow (emulate logger config env) $ - deadline (seconds 5) $ do - xs <- atomically $ do - xs <- readTVar out - unless (length xs >= 2) retry - return xs - names <- forM (reverse xs) $ \x -> - case Json.eitherDecode @Message (LB.fromStrict x) of - Left err -> throwIO $ ErrorCall $ "Message decoding error: " <> err - Right (Message _ _ _ _ (Payload v)) -> return (eventName v) - names `shouldBe` [Just "Wanted", Just "Wanted"] - it "resumes from last index" $ withConfig $ \config -> withPostgresSource $ \table insert source -> do @@ -130,12 +79,6 @@ testEmulator p = describe "emulator" $ do Json.Error str -> Left str Json.Success e -> Right e - eventName v = case v of - Json.Object o -> case KeyMap.lookup "event_name" o of - Just (Json.String name) -> Just name - _ -> Nothing - _ -> Nothing - logger = plainLogger Warn mkEnv sources dests = EnvironmentConfig @@ -167,7 +110,6 @@ testEmulator p = describe "emulator" $ do , d_destination = DestinationFun $ \e -> do atomically $ modifyTVar out (e:) return Nothing - , d_filter = Nothing } return (out, dest) diff --git a/tests/Test/Projector.hs b/tests/Test/Projector.hs deleted file mode 100644 index a4723f3..0000000 --- a/tests/Test/Projector.hs +++ /dev/null @@ -1,52 +0,0 @@ -module Test.Projector (testProjector) where - -import qualified Data.Aeson as Json -import Data.Aeson ((.=)) -import qualified Data.Set as Set -import Data.Text (Text) -import Test.Hspec (Spec, it, describe, shouldBe) - -import Ambar.Emulator.Config (DestinationFilter(..)) -import Ambar.Emulator.Projector (filterVerdict, FilterVerdict(..), Payload(..)) - -testProjector :: Spec -testProjector = - describe "projector filter" $ do - it "no filter delivers everything" $ - filterVerdict Nothing (row "CustomerCreated") `shouldBe` Matched - - it "delivers records whose filter-column value is allowed" $ - filterVerdict (Just orderFilter) (row "OrderCreated") `shouldBe` Matched - - it "skips records whose filter-column value is not allowed" $ - filterVerdict (Just orderFilter) (row "CustomerCreated") `shouldBe` Skipped - - it "fails open (with reason) when the filter column is missing from the record" $ - filterVerdict (Just (filterOn "no_such_column")) (row "OrderCreated") - `shouldBe` FailedOpen "filter column 'no_such_column' is missing from the record" - - it "fails open (with reason) when the filter-column value is not a string" $ - filterVerdict (Just (filterOn "amount")) (row "CustomerCreated") - `shouldBe` FailedOpen "filter column 'amount' has a non-string value" - - it "fails open (with reason) when the filter-column value is null" $ - filterVerdict (Just (filterOn "maybe_null")) (row "CustomerCreated") - `shouldBe` FailedOpen "filter column 'maybe_null' is null" - - it "fails open (with reason) when the record is not an object" $ - filterVerdict (Just orderFilter) (Payload (Json.String "just text")) - `shouldBe` FailedOpen "record is not a JSON object" - where - orderFilter = filterOn "event_name" - - filterOn column = DestinationFilter - { f_column = column - , f_values = Set.fromList ["OrderCreated", "OrderUpdated"] - } - - row eventName = Payload $ Json.object - [ "id" .= (1 :: Int) - , "event_name" .= (eventName :: Text) - , "amount" .= (42 :: Int) - , "maybe_null" .= Json.Null - ] diff --git a/tests/Tests.hs b/tests/Tests.hs index a4405ef..121d111 100644 --- a/tests/Tests.hs +++ b/tests/Tests.hs @@ -4,7 +4,6 @@ import Test.Hspec (hspec, parallel) import Test.Config (testConfig) import Test.Emulator (testEmulator) -import Test.Projector (testProjector) import Test.Transport (testTransport) import Test.Queue (testQueues) import Test.Connector (testConnectors, withDatabases, Databases(..)) @@ -24,7 +23,6 @@ main = hspec $ parallel $ do -- unit tests use the projector library testConfig - testProjector testQueues testTransport testEmulator pcreds