Method test_gridfs_secondary

test/test_gridfs.py:560–572

(self)

        self.assertEqual(b"foo", content)

    def test_gridfs_secondary(self):
        secondary_host, secondary_port = one(self.client.secondaries)
        secondary_connection = self.single_client(
            secondary_host, secondary_port, read_preference=ReadPreference.SECONDARY
        )

        # Should detect it's connected to secondary and not attempt to
        # create index
        fs = gridfs.GridFS(secondary_connection.gfsreplica, "gfssecondarytest")

        # This won't detect secondary, raises error
        with self.assertRaises(NotPrimaryError):
            fs.put(b"foo")

    def test_gridfs_secondary_lazy(self):
        # Should detect it's connected to secondary and not attempt to