Skip to content

adding module on_start to improve how poll callbacks are loaded

Sign in for the full log view
GitHub Actions / Test Results failed May 22, 2024 in 0s

1 fail, 8 skipped, 235 pass in 18s

244 tests  ±0   235 ✅ ±0   18s ⏱️ -1s
  1 suites ±0     8 💤 ±0 
  1 files   ±0     1 ❌ ±0 

Results for commit 3920fca. ± Comparison against earlier commit 9bb5764.

Annotations

Check warning on line 0 in sarracenia.flowcb.accept.testretry_test

See this annotation in the file changed.

@github-actions github-actions / Test Results

test_after_accept (sarracenia.flowcb.accept.testretry_test) failed

tests/junit/test-results.xml [took 0s]
Raw output
KeyError: 'https://TestUsername:TestPassword@options.sendTo.url/Path/File.txt'
caplog = <_pytest.logging.LogCaptureFixture object at 0x7fd83b4b34c0>
mocker = <pytest_mock.plugin.MockerFixture object at 0x7fd83b4b0310>

    @pytest.mark.depends(on=['test___init__'])
    def test_after_accept(caplog, mocker):
        #Set 1 - When random is True, with a message having isRetry
        caplog.clear()
        options = sarracenia.config.default_config()
        options.sendTo = 'http://options.sendTo.url'
        options.logLevel = 'DEBUG'
        worklist = make_worklist()
        worklist.incoming = [make_message()]
        mocker.patch('random.randint', return_value=0)
        testretry = TestRetry(options)
        testretry.after_accept(worklist)
        assert len(worklist.incoming) == 0
        assert "return from testretry after_accept" in caplog.messages
    
    
        #Set 2 - When random is True, with a message having isRetry
        caplog.clear()
        options = sarracenia.config.default_config()
        options.sendTo = 'http://options.sendTo.url'
        options.logLevel = 'DEBUG'
        testretry = TestRetry(options)
        testretry.sendTo = 'http://testretry.sendTo.url'
        testretry.msg_baseUrl_good = 'http://testretry.msg_baseUrl_good.url'
        message_isRetry = make_message(isRetry=True)
        worklist = make_worklist()
        worklist.incoming = [message_isRetry]
        testretry.after_accept(worklist)
        assert len(worklist.incoming) == 0
        assert options.sendTo == 'http://testretry.sendTo.url'
        assert options.details.url.netloc == 'testretry.sendTo.url'
    
    
        #Set 3 - When random is False, there's nothing that gets retried
        caplog.clear()
        options = sarracenia.config.default_config()
        options.sendTo = 'https://TestUsername:TestPassword@options.sendTo.url/Path/File.txt'
        options.component = 'watch'
        options.logLevel = 'DEBUG'
        testretry = TestRetry(options)
        message = make_message()
        worklist = make_worklist()
        worklist.incoming = [message]
        mocker.patch('random.randint', return_value=1)
>       testretry.after_accept(worklist)

tests/sarracenia/flowcb/accept/testretry_test.py:87: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
sarracenia/flowcb/accept/testretry.py:56: in after_accept
    ok, self.o.details = self.o.credentials.get(self.sendTo)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <sarracenia.credentials.CredentialDB object at 0x7fd85c3f5d20>
urlstr = 'https://TestUsername:TestPassword@options.sendTo.url/Path/File.txt'

    def get(self, urlstr):
        """Retrieve a Credential from the DB by urlstr. If the Credential is valid, but not already cached, it will be
        added to the CredentialDB.
    
        Args:
            urlstr (str): credentials as URL string to be parsed.
    
        Returns:
            tuple: containing
                cache_result (bool): ``True`` if the credential was retrieved from the CredentialDB cache, ``False``
                    if it was not in the cache. Note that ``False`` does not imply the Credential or urlstr is
                    invalid.
                credential (sarracenia.credentials.Credential): the Credential
                    object matching the urlstr, ``None`` if urlstr is invalid.
        """
        #logger.debug("CredentialDB get %s" % urlstr)
    
        # already cached
    
        if self.has(urlstr):
            #logger.debug("CredentialDB get in cache %s %s" % (urlstr,self.credentials[urlstr]))
            return True, self.credentials[urlstr]
    
        # create url object if needed
    
        url = urllib.parse.urlparse(urlstr)
    
        # add anonymous default, if necessary.
        if ( 'amqp' in url.scheme ) and \
           ( (url.username == None) or (url.username == '') ):
            urlstr = urllib.parse.urlunparse( ( url.scheme, \
                'anonymous:anonymous@%s' % url.netloc, url.path, None, None, url.port ) )
            url = urllib.parse.urlparse(urlstr)
            if self.isValid(url):
                self.add(urlstr)
                return False, self.credentials[urlstr.replace(':anonymous@','@')]
    
        # resolved from defined credentials
        ok, details = self._resolve(urlstr, url)
        if ok: return True, details
    
        # not found... is it valid ?
        if not self.isValid(url):
            return False, None
    
        # cache it as is... we dont want to validate every time
    
        self.add(urlstr)
>       return False, self.credentials[urlstr]
E       KeyError: 'https://TestUsername:TestPassword@options.sendTo.url/Path/File.txt'

sarracenia/credentials.py:236: KeyError