def test_file_management(self):
        tempDir = tempfile.mkdtemp()

        curOpenFiles = len(os.listdir('/proc/%s/fd' % os.getpid()))

        OPEN_FILE_LIMIT = 200

        if curOpenFiles >= OPEN_FILE_LIMIT:
            os.system("ls -alh /proc/%s/fd" % os.getpid())

        self.assertTrue(curOpenFiles < OPEN_FILE_LIMIT, "Too many open files: %s" % curOpenFiles)

        soft, hard = resource.getrlimit(resource.RLIMIT_OFILE)

        harness = self.getHarness(inMemory=True,
                                  cachePathOverride=tempDir,
                                  maxOpenFiles=15)
        try:
            v1 = harness.newView()
            resource.setrlimit(resource.RLIMIT_OFILE, (curOpenFiles + 30, hard))

            for ix in range(128):
                space = SharedState.Keyspace("TakeHighestIdKeyType",
                                             NativeJson.Json("TestSpace%s" % ix),
                                             1)
                rng = SharedState.KeyRange(space, 0, None, None, True, False)
                v1.subscribe(rng)
                key = SharedState.Key(space, (NativeJson.Json('key%s' % ix),))
                with SharedState.Transaction(v1):
                    v1[key] = NativeJson.Json('value %s' % ix)
        finally:
            time.sleep(0.01)
            harness.teardown()
            resource.setrlimit(resource.RLIMIT_OFILE, (soft, hard))
            try:
                shutil.rmtree(tempDir)
            except:
                pass
dataset/ETHPy150Open ufora/ufora/ufora/distributed/SharedState/SharedState_test.py/TestSharedState.test_file_management/original