def test_pinning_encapsulation(self):
     """Check the pinning getters and setters."""
     self.assertFalse(this_thread_is_pinned())
     pin_this_thread()
     self.assertTrue(this_thread_is_pinned())
     unpin_this_thread()
     self.assertFalse(this_thread_is_pinned())

 def test_decorator_resets(self):
     @use_master
     def check():
         assert this_thread_is_pinned()
     pin_this_thread()
     assert this_thread_is_pinned()
     check()
     assert this_thread_is_pinned()

 def test_decorator_resets(self):
     @use_primary_db
     def check():
         assert this_thread_is_pinned()
     pin_this_thread()
     assert this_thread_is_pinned()
     check()
     assert this_thread_is_pinned()

 def test_decorator(self):
     @use_primary_db
     def check():
         assert this_thread_is_pinned()
     unpin_this_thread()
     assert not this_thread_is_pinned()
     check()
     assert not this_thread_is_pinned()

 def test_context_manager_exception(self):
     unpin_this_thread()
     self.assertFalse(this_thread_is_pinned())
     with self.assertRaises(ValueError):
         with use_master:
             self.assertTrue(this_thread_is_pinned())
             raise ValueError
     self.assertFalse(this_thread_is_pinned())

 def test_decorator(self):
     @use_master
     def check():
         assert this_thread_is_pinned()
     unpin_this_thread()
     assert not this_thread_is_pinned()
     check()
     assert not this_thread_is_pinned()

 def test_context_manager_exception(self):
     unpin_this_thread()
     assert not this_thread_is_pinned()
     with self.assertRaises(ValueError):
         with use_master:
             assert this_thread_is_pinned()
             raise ValueError
     assert not this_thread_is_pinned()

    def test_pinning_encapsulation(self):
        """Check the pinning getters and setters."""
        assert not this_thread_is_pinned(), "Thread started out pinned or this_thread_is_pinned() is broken."

        pin_this_thread()
        assert this_thread_is_pinned(), "pin_this_thread() didn't pin the thread."

        unpin_this_thread()
        assert not this_thread_is_pinned(), "Thread remained pinned after unpin_this_thread()."

 def test_context_manager_exception(self):
     unpin_this_thread()
     try:
         assert not this_thread_is_pinned()
         with use_master:
             assert this_thread_is_pinned()
             raise ValueError
     except ValueError:
         pass
     assert not this_thread_is_pinned()

    def test_decorator_resets(self):

        @use_master
        def check():
            self.assertTrue(this_thread_is_pinned())

        pin_this_thread()
        self.assertTrue(this_thread_is_pinned())
        check()
        self.assertTrue(this_thread_is_pinned())

    def test_decorator(self):

        @use_master
        def check():
            self.assertTrue(this_thread_is_pinned())

        unpin_this_thread()
        self.assertFalse(this_thread_is_pinned())
        check()
        self.assertFalse(this_thread_is_pinned())

    def test_slave_decorator(self):

        @use_slave
        def check():
            self.assertFalse(this_thread_is_pinned())

        pin_this_thread()
        self.assertTrue(this_thread_is_pinned())
        check()
        self.assertTrue(this_thread_is_pinned())

    def test_slave_decorator_resets(self):

        @use_slave
        def check():
            self.assertFalse(this_thread_is_pinned())

        unpin_this_thread()
        self.assertFalse(this_thread_is_pinned())
        check()
        self.assertFalse(this_thread_is_pinned())

 def test_failed_session_auth(self):
     req = RequestFactory().post(
         '/',
         HTTP_AUTHORIZATION='mkt-shared-secret bogus')
     ok_(not self.auth.is_authenticated(req))
     assert not getattr(req, 'amo_user', None)
     ok_(not this_thread_is_pinned())

 def test_request_token_fake(self):
     c = Mock()
     c.key = self.access.key
     c.secret = 'mom'
     ok_(not self.auth.authenticate(
         Request(self.call(client=OAuthClient(c)))))
     ok_(not this_thread_is_pinned())

        def thread2_worker():
            pin_this_thread()
            with use_primary_db:
                orchestrator.release()
                thread2_lock.acquire()

            pinned[2] = this_thread_is_pinned()
            orchestrator.release()

 def test_session_auth_query_disabled(self):
     req = RequestFactory().post('/[email protected],56b6f1a3dd735d962c56'
                                 'ce7d8f46e02ec1d4748d2c00c407d75f0969d08bb'
                                 '9c68c31b3371aa8130317815c89e5072e31bb94b4'
                                 '121c5c165f3515838d4d6c60c4,165d631d3c3045'
                                 '458b4516242dad7ae')
     ok_(not self.auth.is_authenticated(req))
     ok_(not this_thread_is_pinned())

 def test_session_auth_query(self):
     self.create_switch('shared-secret-in-url')
     req = RequestFactory().post('/[email protected],56b6f1a3dd735d962c56'
                                 'ce7d8f46e02ec1d4748d2c00c407d75f0969d08bb'
                                 '9c68c31b3371aa8130317815c89e5072e31bb94b4'
                                 '121c5c165f3515838d4d6c60c4,165d631d3c3045'
                                 '458b4516242dad7ae')
     ok_(self.auth.is_authenticated(req))
     eq_(self.profile.user.pk, req.amo_user.pk)
     ok_(this_thread_is_pinned())

    def process_response(self, request, response):
        if not getattr(request, 'API', False):
            return (super(APIPinningMiddleware, self)
                    .process_response(request, response))

        response['API-Pinned'] = str(this_thread_is_pinned())

        if (request.user and not request.user.is_anonymous() and (
                request.method in ['DELETE', 'PATCH', 'POST', 'PUT'] or
                getattr(response, '_db_write', False))):
            cache.set(self.cache_key(request), 1, PINNING_SECONDS)

        return response

    def process_response(self, request, response):
        if not getattr(request, "API", False):
            return super(APIPinningMiddleware, self).process_response(request, response)

        response["API-Pinned"] = str(this_thread_is_pinned())

        if (
            request.user
            and not request.user.is_anonymous()
            and (request.method in ["DELETE", "PATCH", "POST", "PUT"] or getattr(response, "_db_write", False))
        ):
            cache.set(self.cache_key(request), 1, PINNING_SECONDS)

        return response