[system testing] Implement assertion for expected number of events (#960)

* placeholder in docs

* test-expected-event-count-config.yml in httpjson OK

* add detail to system testing docs

* clarify README.md

* refactor runner

* remove trailing space

* global change SystemTest.ExpectedEventCount to Assert.EventsCount

* Update docs/howto/system_testing.md

Co-authored-by: Jaime Soriano Pastor <[email protected]>

* add explanation to example

* refactor assertion

* assert hits, rather than events

* simplify comparison to bool constant

* WIP adding new test

* rollback changes to generated.log

* change input in test config

* using multiinput as template

* changed to tcp type

* revert hits to tcp

* set hit_count to desired 100

* Update internal/testrunner/runners/system/runner.go

Thanks @andrewkroh

Co-authored-by: Andrew Kroh <[email protected]>

* move declaration inside if block

* wait for expected number of events if hit_count is used

* Apply suggestions from code review

Co-authored-by: Jaime Soriano Pastor <[email protected]>

* working through CI errors

* adding missing fields

---------

Co-authored-by: Jaime Soriano Pastor <[email protected]>
Co-authored-by: Andrew Kroh <[email protected]>
Co-authored-by: Eric Beahan <[email protected]>

Author: 4 people

docs/howto/system_testing.md

@@ -395,6 +395,55 @@ you can use the `input` option to select the stream to test. The first stream
 whose input type matches the `input` value will be tested. By default, the first
 stream declared in the manifest will be tested.
 
+To add an assertion on the number of hits in a given system test, consider this example from the `httpjson/generic` data stream's `test-expected-hit-count-config.yml`, shown below.
+
+```
+input: httpjson
+service: httpjson
+data_stream:
+  vars:
+    data_stream.dataset: httpjson.generic
+    username: test
+    password: test
+    request_url: http://{{Hostname}}:{{Port}}/testexpectedhits/api
+    response_split: |- 
+      target: body.hits
+      type: array
+      keep_parent: false     
+assert:
+  hit_count: 3
+```
+
+The `data_stream.vars.request_url` corresponds to a test-stub path in the `_dev/deploy/docker/files/config.yml` file.
+
+```
+  - path: /testexpectedhits/api
+    methods: ["GET"]
+    request_headers:
+      Authorization:
+        - "Basic dGVzdDp0ZXN0"
+    responses:
+      - status_code: 200
+        headers:
+          Content-Type:
+            - "application/json; charset=utf-8"
+        body: |-
+          {"total":3,"hits":[{"message": "success"},{"message": "success"},{"message": "success"}]}
+```
+
+Handlebar syntax in `httpjson.yml.hbs`
+
+```
+{{#if response_split}}
+response.split: 
+  {{response_split}}
+{{/if}}
+```
+
+inserts the value of `response_split` from the test configuration into the integration, in this case, ensuring the `total.hits[]` array from the test-stub response yields 3 hits.
+
+Returning to `test-expected-hit-count-config.yml`, when `assert.hit_count` is defined and `> 0` the test will assert that the number of hits in the array matches that value and fail when this is not true.
+
 #### Placeholders
 
 The `SERVICE_LOGS_DIR` placeholder is not the only one available for use in a data stream's `test-<test_name>-config.yml` file. The complete list of available placeholders is shown below.

internal/testrunner/runners/system/runner.go

@@ -450,6 +450,11 @@ func (r *runner) runTest(config *testConfig, ctxt servicedeployer.ServiceContext
 
 		var err error
 		docs, err = r.getDocs(dataStream)
+
+		if config.Assert.HitCount > 0 {
+			return len(docs) >= config.Assert.HitCount, err
+		}
+
 		return len(docs) > 0, err
 	}, waitForDataTimeout)
 	if err != nil {
@@ -486,6 +491,11 @@ func (r *runner) runTest(config *testConfig, ctxt servicedeployer.ServiceContext
 		return result.WithError(err)
 	}
 
+	// Check Hit Count within docs, if 0 then it has not been specified
+	if assertionPass, message := assertHitCount(config.Assert.HitCount, docs); !assertionPass {
+		result.FailureMsg = message
+	}
+
 	return result.WithSuccess()
 }
 
@@ -860,6 +870,17 @@ func validateFields(docs []common.MapStr, fieldsValidator *fields.Validator, dat
 	return nil
 }
 
+func assertHitCount(expected int, docs []common.MapStr) (pass bool, message string) {
+	if expected != 0 {
+		observed := len(docs)
+		logger.Debugf("assert hit count expected %d, observed %d", expected, observed)
+		if observed != expected {
+			return false, fmt.Sprintf("observed hit count %d did not match expected hit count %d", observed, expected)
+		}
+	}
+	return true, ""
+}
+
 func (r *runner) selectVariants(variantsFile *servicedeployer.VariantsFile) []string {
 	if variantsFile == nil || variantsFile.Variants == nil {
 		return []string{""} // empty variants file switches to no-variant mode

internal/testrunner/runners/system/test_config.go

@@ -38,6 +38,11 @@ type testConfig struct {
 		Vars common.MapStr `config:"vars"`
 	} `config:"data_stream"`
 
+	Assert struct {
+		// Expected number of hits for a given test
+		HitCount int `config:"hit_count"`
+	} `config:"assert"`
+
 	// NumericKeywordFields holds a list of fields that have keyword
 	// type but can be ingested as numeric type.
 	NumericKeywordFields []string `config:"numeric_keyword_fields"`

test/packages/other/hit_count_assertion/_dev/deploy/docker/docker-compose.yml

@@ -0,0 +1,7 @@
+version: '2.3'
+services:
+  test-hits:
+    image: akroh/stream:v0.0.1
+    volumes:
+      - ./logs:/logs:ro
+    command: log --start-signal=SIGHUP --delay=5s --addr elastic-agent:9999 -p=tcp /logs/generated.log

test/packages/other/hit_count_assertion/_dev/deploy/docker/logs/generated.log

@@ -0,0 +1,100 @@
+ntpd[1001]: kernel time sync enabled utl
+restorecond: : Reset file context quasiarc: liqua
+auditd[5699]: Audit daemon rotating log files
+anacron[5066]: Normal exit ehend
+restorecond: : Reset file context vol: luptat
+heartbeat: : <<eumiu.medium> Processing command: accept
+restorecond: : Reset file context nci: ofdeFin
+auditd[6668]: Audit daemon rotating log files
+anacron[1613]: Normal exit mvolu
+ntpd[2959]: ntpd gelit-r tatno
+anacron[654]: Updated timestamp for job rmagni to sit
+dmd: : <<tenima.very-high> Health state for metric"seq3874.mail.domain" "quid" changed to "fug", reason: "success"
+auditd[2067]: Audit daemon rotating log files
+pm[5969]: <<tquovol.very-high> check_license_validity(), tae
+logrotate: : ALERT exited abnormally with temUten
+sshd: : <<dun.medium> error: Bind to port Duisau on psum failed: failure
+configd: : <<end.medium> itaut@rveli: command: accept
+authd: : <<luptat.low> authd_signal_handler(), quam
+xinetd[6547]: Started working: onproide available services
+logrotate: : ALERT exited abnormally with tfug
+heartbeat: : <<urE.medium> Processing command: deny
+rsyslogd: : Warning: rehe
+sshd: : <<stiae.medium> error: Bind to port erc on amqu failed: unknown
+ntpd[4515]: ntpd emp-r aperia
+restorecond: : Reset file context run: vol
+logrotate: : ALERT exited abnormally with mporain
+heartbeat: : <<mpori.very-high> connect: atu
+cmd: : <<texp.medium> cmd starting adeseru
+cli[7108]: <<-uam.low> tmo@::fficiade:10.2.53.125 : CLI launched
+pm[7061]: <<ihilmo.very-high> ntpd will start in tlabo
+poller[795]: <<oluptate.low> Querying content system for job results.
+runner[6134]: <<edo.very-high> Processing command: allow
+epmd: : epmd: epmd running orpor
+runner[602]: <<emvel.very-high> Failed to exec olup
+shutdown[2807]: shutting down non
+configd: : <<ugiatnu.high> sperna@sintocc: command: cancel
+auditd[2986]: Audit daemon rotating log files
+configd: : <<ccaecat.medium> CREATE onsequ
+auditd[1243]: Audit daemon rotating log files
+xinetd[6599]: Started working: naal available services
+xinetd[5850]: Started working: rQu available services
+heartbeat: : <<boree.low> queips: undefined symbol: ncidi
+authd: : <<olor.very-high> authd_close(): npr
+anacron[6373]: Anacron 1.3962 started on epre
+cli[3979]: <<-iduntu.medium> [email protected] : Processing command accept
+cmd: : <<amc.medium> cmd starting isiuta
+sshd[5227]: dutp(psaquaea:taevita): pam_putenv: ameiusm
+ccd: : <<olab.low> Device elitse6672.internal.localdomain: mquisno
+runner[1859]: <<tasnulap.high> Failed to exec umSe
+shutdown[6110]: shutting down itau
+sshd[2415]: PAM lorsita more authentication failure; dolore
+rsyslogd: : Warning: tio
+cli[802]: <<-gnaaliqu.very-high> velillu@::cteturad:10.18.204.87 : Processing a secure command...
+heartbeat: : <<reprehe.high> connect: inimveni
+authd: : <<litani.low> authd_close(): psumqu
+runner[2558]: <<icabo.high> Failed to exec edquiac
+anacron[4538]: Updated timestamp for job remips to uisaute
+auditd[6837]: Audit daemon rotating log files
+pm[1493]: <<etdolor.high> print_msg(), dic
+configd: : <<avolupt.low> Device "itation4168.api.domain" completed command(s) accept ;; CPL generated by Visual Policy Manager: isciv ;rroqu ; nofd ; dipisci
+epmd: : epmd: invalid packet size (mquae)
+runner[429]: <<corpori.very-high> File reading failed
+shutdown[7595]: shutting down emqu
+heartbeat: : <<leumiur.low> The HB command is accept
+authd: : <<est.very-high> authd_signal_handler(), isetquas
+authd: : <<psaqua.medium> authd_signal_handler(), gnaal
+logrotate: : ALERT exited abnormally with voluptas
+ntpd[627]: ntpd exiting on signal orin
+restorecond: : Reset file context ecillu: mmodoc
+cli[1140]: <<-abore.high> [email protected] : Processing command: deny
+sshd: : bad username mquisn
+ntpd[1313]: ntpd derit-r orese
+ccd: : <<leumiur.medium> Device Communication Daemon online
+rsyslogd: : Warning: moles
+restorecond: : Reset file context olup: aco
+shutdown[609]: shutting down ser
+ntpd[2991]: ntpd orinrep-r quiavol
+dmd: : <<quin.medium> inserted device id = sBonor2001.www5.example and serial number = amc into DB
+ccd: : <<ame.very-high> ccd_handle_read_failure(), uid
+cmd: : <<scivel.high> cmd starting lmolesti
+dmd: : <<emaperia.high> inserted device id = ersp6625.internal.domain and serial number = seq into DB
+cmd: : <<tanimid.medium> cmd starting uipexe
+heartbeat: : <<ore.low> The HB command is cancel
+anacron[7360]: Normal exit tperspic
+dmd: : <<ict.very-high> Filter on (tetura) things. riosamni
+ccd: : <<umetMa.low> Device eleumiu2454.api.local: tat
+schedulerd: : <<lumqu.very-high> System time changed, recomputing job run times.
+xinetd[3450]: Started working: aconsequ available services
+authd: : <<sequat.high> handle_authd unknown message =utemvel
+rsyslogd: : Warning: iusm
+ntpd[16]: time reset stquido
+ccd: : <<aaliq.high> Device olu5333.www.domain: orumSe
+anacron[80]: Normal exit ici
+ntpd[7612]: kernel time sync enabled nturmag
+cli[7128]: eseruntm(lpaquiof:oloreeu): pam_putenv: olor
+schedulerd: : <<ici.very-high> Executing Job "tquo" execution iatnu
+logrotate: : ALERT exited abnormally with ntut
+poller[7151]: <<ess.high> Querying content system for job results.
+ntpd[2314]: ntpd litanim-r rQuisaut
+heartbeat: : <<metco.high> Processing command: block

test/packages/other/hit_count_assertion/changelog.yml

@@ -0,0 +1,6 @@
+# newer versions go on top
+- version: "999.999.999"
+  changes:
+    - description: initial release
+      type: enhancement # can be one of: enhancement, bugfix, breaking-change
+      link: https://github.com/elastic/elastic-package/pull/960

test/packages/other/hit_count_assertion/data_stream/test/_dev/test/system/test-hits-config.yml

@@ -0,0 +1,10 @@
+input: tcp
+service: test-hits
+service_notify_signal: SIGHUP
+vars: ~
+data_stream:
+  vars:
+    tcp_host: 0.0.0.0
+    tcp_port: 9999
+assert:
+  hit_count: 100

test/packages/other/hit_count_assertion/data_stream/test/agent/stream/hits.yml.hbs

@@ -0,0 +1,21 @@
+host: "{{tcp_host}}:{{tcp_port}}"
+tags:
+{{#each tags as |tag i|}}
+ - {{tag}}
+{{/each}}
+fields_under_root: true
+fields:
+    observer:
+        vendor: Test
+        product: test
+        type: test
+{{#contains tags "forwarded"}}
+publisher_pipeline.disable_host: true
+{{/contains}}
+
+processors:
+- add_locale: ~
+- add_fields:
+    target: ''
+    fields:
+        ecs.version: 1.7.0

test/packages/other/hit_count_assertion/data_stream/test/agent/stream/stream.yml.hbs

@@ -0,0 +1,25 @@
+paths:
+{{#each paths as |path i|}}
+  - {{path}}
+{{/each}}
+exclude_files: [".gz$"]
+tags:
+{{#each tags as |tag i|}}
+ - {{tag}}
+{{/each}}
+fields_under_root: true
+fields:
+    observer:
+        vendor: Test
+        product: Test
+        type: test
+{{#contains tags "forwarded"}}
+publisher_pipeline.disable_host: true
+{{/contains}}
+
+processors:
+- add_locale: ~
+- add_fields:
+    target: ''
+    fields:
+        ecs.version: 1.6.0

test/packages/other/hit_count_assertion/data_stream/test/elasticsearch/ingest_pipeline/default.yml

@@ -0,0 +1,12 @@
+---
+description: Test Pipeline
+
+processors:
+  # ECS event.ingested
+  - set:
+        field: event.ingested
+        value: '{{_ingest.timestamp}}'
+on_failure:
+  - append:
+        field: error.message
+        value: "{{ _ingest.on_failure_message }}"