(self, dis, reg, key='KEY', cert='CERT')
| 127 | @patch('celery.security.register_auth') |
| 128 | @patch('celery.security._disable_insecure_serializers') |
| 129 | def test_setup_registry_complete(self, dis, reg, key='KEY', cert='CERT'): |
| 130 | calls = [0] |
| 131 | |
| 132 | def effect(*args): |
| 133 | try: |
| 134 | m = Mock() |
| 135 | m.read.return_value = 'B' if calls[0] else 'A' |
| 136 | return m |
| 137 | finally: |
| 138 | calls[0] += 1 |
| 139 | |
| 140 | self.app.conf.task_serializer = 'auth' |
| 141 | self.app.conf.accept_content = ['auth'] |
| 142 | with conftest.open(side_effect=effect): |
| 143 | with patch('celery.security.registry') as registry: |
| 144 | store = Mock() |
| 145 | self.app.setup_security(['json'], key, None, cert, store) |
| 146 | dis.assert_called_with(['json']) |
| 147 | reg.assert_called_with('A', None, 'B', store, 'sha256', 'json') |
| 148 | registry._set_default_serializer.assert_called_with('auth') |
| 149 | |
| 150 | def test_security_conf(self): |
| 151 | self.app.conf.task_serializer = 'auth' |
nothing calls this directly
no test coverage detected