<commit_before>import io
import hashlib

import synapse.axon as s_axon
import synapse.daemon as s_daemon
import synapse.lib.heap as s_heap
import synapse.telepath as s_telepath
import synapse.lib.service as s_service

from synapse.tests.common import *

craphash = 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'
asdfhash = '6a204bd89f3c8348afd5c77c717a097a'

class AxonTest(SynTest):

    def test_axon_basics(self):
        with self.getTestDir() as axondir:

            axon = s_axon.Axon(axondir)

            self.assertFalse( axon.has('md5',craphash) )
            self.assertFalse( axon.has('md5',asdfhash) )

            iden0 = axon.alloc(8)

            self.assertIsNotNone( axon.chunk(iden0,b'asdfasdf') )

            self.assertTrue( axon.has('md5',asdfhash) )
            self.assertFalse( axon.has('md5',craphash) )

            byts = b''.join( axon.bytes('md5',asdfhash) )

            self.assertEqual(byts,b'asdfasdf')

            axon.fini()

            axon = s_axon.Axon(axondir)

            self.assertTrue( axon.has('md5',asdfhash) )
            self.assertFalse( axon.has('md5',craphash) )

            byts = b''.join( axon.bytes('md5',asdfhash) )

            self.assertEqual(byts,b'asdfasdf')

            self.assertIsNone(axon.wants('md5', asdfhash, 8))
            self.assertIsNotNone(axon.wants('md5', craphash, 8))

            axon.fini()

    def test_axon_sync(self):

        with self.getTestDir() as axondir:

            byts = os.urandom(128)
            bytsmd5 = hashlib.md5(byts).hexdigest()

            axon = s_axon.Axon(axondir,syncsize=64)

            iden = axon.alloc(128)
            for chnk in chunks(byts,10):
                blob = axon.chunk(iden,chnk)

            self.assertIsNotNone(blob)

            self.assertTrue( axon.has('md5',bytsmd5) )

            axon.fini()

    def test_axon_host(self):

        self.thisHostMustNot(platform='windows')

        with self.getTestDir() as datadir:
            open(os.path.join(datadir, 'foo'), 'w').write('useless file to skip')
<commit_msg>eliminate leaked fd race in unit test<commit_after>import io
import hashlib

import synapse.axon as s_axon
import synapse.daemon as s_daemon
import synapse.lib.heap as s_heap
import synapse.telepath as s_telepath
import synapse.lib.service as s_service

from synapse.tests.common import *

craphash = 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'
asdfhash = '6a204bd89f3c8348afd5c77c717a097a'

class AxonTest(SynTest):

    def test_axon_basics(self):
        with self.getTestDir() as axondir:

            axon = s_axon.Axon(axondir)

            self.assertFalse( axon.has('md5',craphash) )
            self.assertFalse( axon.has('md5',asdfhash) )

            iden0 = axon.alloc(8)

            self.assertIsNotNone( axon.chunk(iden0,b'asdfasdf') )

            self.assertTrue( axon.has('md5',asdfhash) )
            self.assertFalse( axon.has('md5',craphash) )

            byts = b''.join( axon.bytes('md5',asdfhash) )

            self.assertEqual(byts,b'asdfasdf')

            axon.fini()

            axon = s_axon.Axon(axondir)

            self.assertTrue( axon.has('md5',asdfhash) )
            self.assertFalse( axon.has('md5',craphash) )

            byts = b''.join( axon.bytes('md5',asdfhash) )

            self.assertEqual(byts,b'asdfasdf')

            self.assertIsNone(axon.wants('md5', asdfhash, 8))
            self.assertIsNotNone(axon.wants('md5', craphash, 8))

            axon.fini()

    def test_axon_sync(self):

        with self.getTestDir() as axondir:

            byts = os.urandom(128)
            bytsmd5 = hashlib.md5(byts).hexdigest()

            axon = s_axon.Axon(axondir,syncsize=64)

            iden = axon.alloc(128)
            for chnk in chunks(byts,10):
                blob = axon.chunk(iden,chnk)

            self.assertIsNotNone(blob)

            self.assertTrue( axon.has('md5',bytsmd5) )

            axon.fini()

    def test_axon_host(self):

        self.thisHostMustNot(platform='windows')

        with self.getTestDir() as datadir:

            with open(os.path.join(datadir,'foo'),'w') as fd:
                fd.write('useless file to skip')

{'fraction_non_alphanumeric': 0.11993243243243243, 'fraction_numerical': 0.02364864864864865, 'mean_word_length': 2.358995137763371, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7361170', 'n_tokens_mistral': 1557, 'n_tokens_neox': 1395, 'n_words': 232}