package Ġstreams Ċ Ċ import Ġorg . junit . Test Ċ import Ġstreams . service . Topic Type Ċ import Ġstreams . service . Topic ValidationException Ċ import Ġkotlin . test . assertEquals Ċ import Ġkotlin . test . assertFalse Ċ import Ġkotlin . test . assertTrue Ċ Ċ class ĠStreams Sink Configuration Test Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġdefault Db Name Ġ= Ġ" neo 4 j " ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` should Ġmanage Ġonly Ġtopics Ġfor Ġdefault Ġdb `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġtopic Key Ġ= Ġ" streams . sink . topic . cy pher . my Topic " ĊĠĠĠĠĠĠĠ Ġval Ġtopic Value Ġ= Ġ" MERGE Ġ( n : Label { Ġid : Ġevent . id Ġ}) " ĊĠĠĠĠĠĠĠ Ġval Ġtopic Key Neo Ġ= Ġ" streams . sink . topic . cy pher . my Topic Neo . to . neo 4 j " ĊĠĠĠĠĠĠĠ Ġval Ġtopic Value Neo Ġ= Ġ" MERGE Ġ( n : Neo 4 j { Ġid : Ġevent . id Ġ}) " ĊĠĠĠĠĠĠĠ Ġval Ġtopic Key Foo Ġ= Ġ" streams . sink . topic . cy pher . my Topic Foo . to . foo " ĊĠĠĠĠĠĠĠ Ġval Ġtopic Value Foo Ġ= Ġ" MERGE Ġ( n : Foo { Ġid : Ġevent . id Ġ}) " ĊĠĠĠĠĠĠĠ Ġval Ġconfig Ġ= Ġmap Of ( topic Key Ġto Ġtopic Value , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtopic Key Neo Ġto Ġtopic Value Neo , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtopic Key Foo Ġto Ġtopic Value Foo ) ĊĠĠĠĠĠĠĠ Ġval Ġstreams Sink Conf Ġ= ĠStreams Sink Configuration . from ( config , Ġdefault Db Name , Ġis Default Db Ġ= Ġtrue ) ĊĠĠĠĠĠĠĠ Ġval Ġcy pher Topics Ġ= Ġstreams Sink Conf . topics . as Map ()[ Topic Type . CY P HER ] Ġas ĠMap < String , ĠString > ĊĠĠĠĠĠĠĠ ĠassertEquals ( map Of (" my Topic " Ġto Ġtopic Value , Ġ" my Topic Neo " Ġto Ġtopic Value Neo ), Ġcy pher Topics ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` should Ġmanage Ġonly Ġtopics Ġfor Ġnon Ġdefault Ġdb `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġtopic Key Ġ= Ġ" streams . sink . topic . cy pher . my Topic " ĊĠĠĠĠĠĠĠ Ġval Ġtopic Value Ġ= Ġ" MERGE Ġ( n : Label { Ġid : Ġevent . id Ġ}) " ĊĠĠĠĠĠĠĠ Ġval Ġtopic Key Neo Ġ= Ġ" streams . sink . topic . cy pher . my Topic Neo . to . neo 4 j " ĊĠĠĠĠĠĠĠ Ġval Ġtopic Value Neo Ġ= Ġ" MERGE Ġ( n : Neo 4 j { Ġid : Ġevent . id Ġ}) " ĊĠĠĠĠĠĠĠ Ġval Ġtopic Key Foo Ġ= Ġ" streams . sink . topic . cy pher . my Topic Foo . to . foo " ĊĠĠĠĠĠĠĠ Ġval Ġtopic Value Foo Ġ= Ġ" MERGE Ġ( n : Foo { Ġid : Ġevent . id Ġ}) " ĊĠĠĠĠĠĠĠ Ġval Ġconfig Ġ= Ġmap Of ( topic Key Ġto Ġtopic Value , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtopic Key Neo Ġto Ġtopic Value Neo , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtopic Key Foo Ġto Ġtopic Value Foo ) ĊĠĠĠĠĠĠĠ Ġval Ġstreams Sink Conf Ġ= ĠStreams Sink Configuration . from ( config , Ġ" foo ", Ġis Default Db Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠ Ġval Ġcy pher Topics Ġ= Ġstreams Sink Conf . topics . as Map ()[ Topic Type . CY P HER ] Ġas ĠMap < String , ĠString > ĊĠĠĠĠĠĠĠ ĠassertEquals ( map Of (" my Topic Foo " Ġto Ġtopic Value Foo ), Ġcy pher Topics ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġshould Return Default Configuration () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġdefault Ġ= ĠStreams Sink Configuration () ĊĠĠĠĠĠĠĠ Ġtest Default Conf ( default ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġshould Return Configuration From Map () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġtopic Ġ= Ġ" topic - neo " ĊĠĠĠĠĠĠĠ Ġval Ġcd ct op ic Ġ= Ġ" cd ct op ic " ĊĠĠĠĠĠĠĠ Ġval Ġtopic Key Ġ= Ġ" streams . sink . topic . cy pher .$ topic " ĊĠĠĠĠĠĠĠ Ġval Ġtopic Value Ġ= Ġ" MERGE Ġ( n : Label { Ġid : Ġevent . id Ġ}) Ġ" ĊĠĠĠĠĠĠĠ Ġval Ġcustom Label Ġ= Ġ" Custom Label " ĊĠĠĠĠĠĠĠ Ġval Ġcustom Id Ġ= Ġ" custom Id " ĊĠĠĠĠĠĠĠ Ġval Ġap oc Timeout Ġ= Ġ" 1 0 0 0 0 " ĊĠĠĠĠĠĠĠ Ġval Ġap oc Interval Ġ= Ġ" 2 0 0 0 " ĊĠĠĠĠĠĠĠ Ġval Ġcluster Only Ġ= Ġ" true " ĊĠĠĠĠĠĠĠ Ġval Ġwrite able Instance Interval Ġ= Ġ" 9 9 " ĊĠĠĠĠĠĠĠ Ġval Ġpoll Inter v all Ġ= Ġ" 1 0 0 " ĊĠĠĠĠĠĠĠ Ġval Ġconfig Ġ= Ġmap Of ( topic Key Ġto Ġtopic Value , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" streams . sink . enabled " Ġto Ġ" false ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" streams . sink . topic . cdc . source Id " Ġto Ġcd ct op ic , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" streams . sink . topic . cdc . source Id . label Name " Ġto Ġcustom Label , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" streams . check . ap oc . timeout " Ġto Ġap oc Timeout , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" streams . check . ap oc . interval " Ġto Ġap oc Interval , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" streams . sink . topic . cdc . source Id . id Name " Ġto Ġcustom Id , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" streams . cluster . only " Ġto Ġcluster Only , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" streams . sink . poll . interval " Ġto Ġpoll Inter v all , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" streams . check . write able . instance . interval " Ġto Ġwrite able Instance Interval ) ĊĠĠĠĠĠĠĠ Ġval Ġstreams Sink Conf Ġ= ĠStreams Sink Configuration . from ( config , Ġdefault Db Name , Ġis Default Db Ġ= Ġtrue ) ĊĠĠĠĠĠĠĠ Ġtest From Conf ( streams Sink Conf , Ġtopic , Ġtopic Value ) ĊĠĠĠĠĠĠĠ ĠassertFalse Ġ{ Ġstreams Sink Conf . enabled Ġ} ĊĠĠĠĠĠĠĠ ĠassertTrue Ġ{ Ġstreams Sink Conf . cluster Only Ġ} ĊĠĠĠĠĠĠĠ ĠassertEquals ( set Of ( cd ct op ic ), Ġstreams Sink Conf . topics . as Map ()[ Topic Type . CDC _ SOURCE _ ID ]) ĊĠĠĠĠĠĠĠ ĠassertEquals ( custom Label , Ġstreams Sink Conf . source Id Strategy Config . label Name ) ĊĠĠĠĠĠĠĠ ĠassertEquals ( custom Id , Ġstreams Sink Conf . source Id Strategy Config . id Name ) ĊĠĠĠĠĠĠĠ ĠassertEquals ( poll Inter v all . to Long (), Ġstreams Sink Conf . poll Interval ) ĊĠĠĠĠĠĠĠ ĠassertEquals ( ap oc Timeout . to Long (), Ġstreams Sink Conf . check Ap oc Timeout ) ĊĠĠĠĠĠĠĠ ĠassertEquals ( ap oc Interval . to Long (), Ġstreams Sink Conf . check Ap oc Interval ) ĊĠĠĠĠĠĠĠ ĠassertEquals ( write able Instance Interval . to Long (), Ġstreams Sink Conf . check Write able Instance Interval ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġshould Return Configuration From Map With Non LowerCase Db Name () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġtopic Ġ= Ġ" my topic " ĊĠĠĠĠĠĠĠ Ġval Ġtopic Key Ġ= Ġ" streams . sink . topic . cy pher .$ topic . to . non LowerCase Db " ĊĠĠĠĠĠĠĠ Ġval Ġtopic Value Ġ= Ġ" MERGE Ġ( n : Label { Ġid : Ġevent . id Ġ}) " ĊĠĠĠĠĠĠĠ Ġval Ġconfig Ġ= Ġmap Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtopic Key Ġto Ġtopic Value , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" streams . sink . enabled " Ġto Ġ" false ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" streams . sink . enabled . to . non LowerCase Db " Ġto Ġ" true ") ĊĠĠĠĠĠĠĠ Ġval Ġstreams Sink Conf Ġ= ĠStreams Sink Configuration . from ( config , Ġ" non lowercase db ", Ġis Default Db Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠ ĠassertFalse Ġ{ Ġstreams Sink Conf . enabled Ġ} ĊĠĠĠĠĠĠĠ ĠassertEquals ( topic Value , Ġstreams Sink Conf . topics . cy pher Topics [ topic ]) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ( expected Ġ= ĠTopic ValidationException :: class ) ĊĠĠĠ Ġfun Ġshould Fail With Cross Defined Topics () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġtopic Ġ= Ġ" topic - neo " ĊĠĠĠĠĠĠĠ Ġval Ġtopic Key Ġ= Ġ" streams . sink . topic . cy pher .$ topic " ĊĠĠĠĠĠĠĠ Ġval Ġtopic Value Ġ= Ġ" MERGE Ġ( n : Label { Ġid : Ġevent . id Ġ}) Ġ" ĊĠĠĠĠĠĠĠ Ġval Ġconfig Ġ= Ġmap Of ( topic Key Ġto Ġtopic Value , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" streams . sink . topic . pattern . node . node Pattern Topic " Ġto Ġ" User { ! userId , name , surname , address . city }", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" streams . sink . enabled " Ġto Ġ" false ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" streams . sink . topic . cdc . source Id " Ġto Ġtopic ) ĊĠĠĠĠĠĠĠ ĠStreams Sink Configuration . from ( config , Ġdefault Db Name , Ġis Default Db Ġ= Ġtrue ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ( expected Ġ= ĠTopic ValidationException :: class ) ĊĠĠĠ Ġfun Ġshould Fail With Cross Defined CD CT op ics () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġtopic Ġ= Ġ" topic - neo " ĊĠĠĠĠĠĠĠ Ġval Ġconfig Ġ= Ġmap Of (" streams . sink . enabled " Ġto Ġ" false ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" streams . sink . topic . cdc . source Id " Ġto Ġtopic , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" streams . sink . topic . cdc . schema " Ġto Ġtopic ) ĊĠĠĠĠĠĠĠ ĠStreams Sink Configuration . from ( config , Ġdefault Db Name , Ġis Default Db Ġ= Ġtrue ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġfun Ġtest Default Conf ( default : ĠStreams Sink Configuration ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( empty Map (), Ġdefault . topics . cy pher Topics ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġfun Ġtest From Conf ( streams Config : ĠStreams Sink Configuration , Ġtopic : ĠString , Ġtopic Value : ĠString ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( 1 , Ġstreams Config . topics . cy pher Topics . size ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( topic Value , Ġstreams Config . topics . cy pher Topics [ topic ]) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ Ċ }