title: Mocking Eventbridge event pattern matching with moto
body: <p>Is it possible to mock Eventbridge pattern matching, confirmed by the receipt of a pattern- matched message into an SQS queue ?</p>
  <p>Asking as I have a boto3/moto script (below) which attempts to do this and which I believe <em>should</em> work, but .. it's not :/</p>
  <p>I <em>think</em> this probably falls under the title &quot;complex mocked AWS workflows never behave exactly the way they do in the cloud&quot;, but it's possible I've made a mistake somewhere.</p>
  <p>Anyone know if testing of Eventbridge pattern matching should be possible in this way (or not) or if I am doing something wrong ?</p>
  <p>TIA</p>
  <pre class="lang-py prettyprint-override"><code>import boto3, json, unittest
  
  from datetime import datetime
  
  from moto import mock_events, mock_sqs
  
  QueueName, RuleName = &quot;hello-queue&quot;, &quot;hello-rule&quot;
  
  RuleTargetId, RuleEnabled = &quot;hello-rule-target&quot;, &quot;ENABLED&quot;
  
  EventSource, EventDetailType, EventResource = &quot;hello-event-source&quot;, &quot;hello-event-detail-type&quot;, &quot;event-resource&quot;
  
  @mock_events
  @mock_sqs
  class HelloTest(unittest.TestCase):
      
      def setUp(self):
          self.sqs, self.events = (boto3.client(&quot;sqs&quot;),
                                   boto3.client(&quot;events&quot;))
          statement=[{&quot;Effect&quot;: &quot;Allow&quot;,
                      &quot;Principal&quot;: {&quot;Service&quot;: &quot;events.amazonaws.com&quot;},
                      &quot;Action&quot;: &quot;sqs:SendMessage&quot;,
                      &quot;Resource&quot;: &quot;*&quot;}]
          policy=json.dumps({&quot;Version&quot;: &quot;2012-10-17&quot;,
                             &quot;Statement&quot;: [statement]})
          queue=self.sqs.create_queue(QueueName=QueueName,
                                      Attributes={&quot;Policy&quot;: policy})
          self.queueurl=queue[&quot;QueueUrl&quot;]
          queueattrs=self.sqs.get_queue_attributes(QueueUrl=self.queueurl)
          self.queuearn=queueattrs[&quot;Attributes&quot;][&quot;QueueArn&quot;]
          eventpattern={&quot;source&quot;: [EventSource],
                        &quot;detail&quot;: {&quot;hello&quot;: [&quot;world&quot;]}}
          self.rule=self.events.put_rule(Name=RuleName,
                                         State=RuleEnabled,
                                         EventPattern=json.dumps(eventpattern))
          self.events.put_targets(Rule=RuleName,
                                  Targets=[{&quot;Id&quot;: RuleTargetId,
                                            &quot;Arn&quot;: self.queuearn}])
  
      # test that simple SQS send/receive works in moto
          
      def test_sqs_message(self):
          reqbody=json.dumps({&quot;hello&quot;: &quot;world&quot;})
          self.sqs.send_message(QueueUrl=self.queueurl,
                                MessageBody=reqbody)
          sqsresp=self.sqs.receive_message(QueueUrl=self.queueurl)
          self.assertTrue(&quot;Messages&quot; in sqsresp)
          messages=sqsresp[&quot;Messages&quot;]
          self.assertTrue(len(messages), 1)
          message=messages.pop()
          body=json.loads(message[&quot;Body&quot;])
          self.assertTrue(&quot;hello&quot; in body)
  
      # test that more complex Eventbridge+SQS workflow works
          
      def test_eventbridge_message(self):
          detail=json.dumps({&quot;hello&quot;: &quot;world&quot;})
          entry={&quot;Time&quot;: datetime.utcnow().strftime(&quot;%Y-%m-%dT%H:%M:%SZ&quot;),
                 &quot;Source&quot;: EventSource,
                 &quot;Resources&quot;: [EventResource],
                 &quot;DetailType&quot;: EventDetailType,
                 &quot;Detail&quot;: detail}
          self.events.put_events(Entries=[entry])
          sqsresp=self.sqs.receive_message(QueueUrl=self.queueurl)
          # self.assertTrue(&quot;Messages&quot; in sqsresp) # &lt;--- FAILS HERE
    
      def tearDown(self):
          self.sqs.delete_queue(QueueUrl=self.queueurl)
          self.events.remove_targets(Rule=RuleName,
                                     Ids=[RuleTargetId])
          self.events.delete_rule(Name=RuleName)
          
  if __name__==&quot;__main__&quot;:
      unittest.main()
  </code></pre>