STUDY METHODS SOURCE EXCERPTS; not a standalone executable.
Source identifier: shared-original-collector

--- Original lines 145-184 ---
    def exact(self, label, app, sql, expected):
        start = len(self.calls)
        try:
            observed = self.query[app](sql).strip()
            status = 'PASS' if observed == str(expected) else 'FAIL'
            error = None
        except Exception as e:
            observed, status, error = None, 'UNKNOWN', str(e)
        self.extra.append(dict(label=label, status=status, observed=observed, expected=expected,
                               error=error, call_indices=list(range(start + 1, len(self.calls) + 1))))

    def exact_checks(self):
        i = self.instance
        q = lambda k: literal(i[k])
        customer = 'SELECT ID FROM CONTACTS WHERE CONTACT_SERVICE=\'customer\' AND DISPLAY_NAME=' + q('company')
        inv = 'SELECT ID FROM SALES_INVOICES WHERE CUSTOMER_ID IN (' + customer + ')'
        self.exact('Exactly one customer', 'bigcapital', 'SELECT COUNT(*) FROM (' + customer + ') x', 1)
        self.exact('Exactly two customer invoices', 'bigcapital', 'SELECT COUNT(*) FROM (' + inv + ') x', 2)
        for n in [1, 2]:
            selected = inv + ' AND INVOICE_DATE=' + literal(i['dates']['invoice' + str(n)])
            self.exact('Invoice %d has exactly one line' % n, 'bigcapital',
                       "SELECT COUNT(*) FROM ITEMS_ENTRIES WHERE REFERENCE_TYPE='SaleInvoice' AND REFERENCE_ID IN (" + selected + ')', 1)
            self.exact('Invoice %d balance' % n, 'bigcapital', 'SELECT COUNT(*) FROM SALES_INVOICES WHERE ID IN (' + selected + ') AND ABS(BALANCE-' + str(0 if n == 1 else i['phase2_amount']) + ')<0.01', 1)
        payment = 'SELECT ID FROM PAYMENT_RECEIVES WHERE CUSTOMER_ID IN (' + customer + ')'
        self.exact('Exactly one customer payment', 'bigcapital', 'SELECT COUNT(*) FROM (' + payment + ') x', 1)
        self.exact('Exactly one payment allocation', 'bigcapital', 'SELECT COUNT(*) FROM PAYMENT_RECEIVES_ENTRIES WHERE PAYMENT_RECEIVE_ID IN (' + payment + ')', 1)
        self.exact('Payment allocated only to this customer milestone 1', 'bigcapital', 'SELECT COUNT(*) FROM PAYMENT_RECEIVES_ENTRIES WHERE PAYMENT_RECEIVE_ID IN (' + payment + ') AND INVOICE_ID IN (' + inv + ' AND INVOICE_DATE=' + literal(i['dates']['invoice1']) + ') AND ABS(PAYMENT_AMOUNT-' + str(i['phase1_amount']) + ')<0.01', 1)
        journal = 'SELECT ID FROM MANUAL_JOURNALS WHERE DESCRIPTION=' + q('journal_memo')
        self.exact('Exactly one named journal', 'bigcapital', 'SELECT COUNT(*) FROM (' + journal + ') x', 1)
        self.exact('Exactly two journal lines', 'bigcapital', 'SELECT COUNT(*) FROM MANUAL_JOURNALS_ENTRIES WHERE MANUAL_JOURNAL_ID IN (' + journal + ')', 2)
        self.exact('Journal lines have no opposite-side amount', 'bigcapital', 'SELECT COUNT(*) FROM MANUAL_JOURNALS_ENTRIES e JOIN ACCOUNTS a ON a.ID=e.ACCOUNT_ID WHERE e.MANUAL_JOURNAL_ID IN (' + journal + ') AND ((a.NAME=' + q('deferred_account') + ' AND ABS(COALESCE(e.CREDIT,0))<0.01) OR (a.NAME=' + q('income_account') + ' AND ABS(COALESCE(e.DEBIT,0))<0.01))', 2)
        event = 'SELECT e.id FROM pretixbase_event e JOIN pretixbase_organizer o ON o.id=e.organizer_id WHERE e.slug=' + q('event_slug') + ' AND o.slug=' + q('organizer')
        self.exact('Exactly one named event', 'pretix', 'SELECT COUNT(*) FROM (' + event + ') x', 1)
        for table in ['pretixbase_item', 'pretixbase_quota', 'pretixbase_voucher']:
            self.exact('Exactly one event ' + table, 'pretix', 'SELECT COUNT(*) FROM ' + table + ' WHERE event_id IN (' + event + ')', 1)
        self.exact('Exactly one quota-product association', 'pretix', 'SELECT COUNT(*) FROM pretixbase_quota_items WHERE quota_id IN (SELECT id FROM pretixbase_quota WHERE event_id IN (' + event + '))', 1)
        self.exact('Event starts at declared UTC instant', 'pretix', "SELECT COUNT(*) FROM pretixbase_event WHERE id IN (" + event + ") AND date_from=" + literal(i['dates']['event'] + ' ' + i['event_start_time'] + '+00') + '::timestamptz', 1)
        ws = self.module.get_twenty_workspace_schema().replace('"', '""')
        for name, body in [('collection_task', 'collection_body'), ('invitation_task', 'invitation_body')]:
            self.exact('Full task body: ' + i[name], 'twenty', 'SELECT COUNT(*) FROM "' + ws + '".task WHERE title=' + q(name) + ' AND "deletedAt" IS NULL AND regexp_replace(trim("bodyV2Markdown"), \'[[:space:]]+\', \' \', \'g\')=' + literal(' '.join(i[body].split())), 1)

