<commit_before>#
#    ICRAR - International Centre for Radio Astronomy Research
#    (c) UWA - The University of Western Australia, 2015
#    Copyright by UWA (in the framework of the ICRAR)
#    All rights reserved
#
#    This library is free software; you can redistribute it and/or
#    modify it under the terms of the GNU Lesser General Public
#    License as published by the Free Software Foundation; either
#    version 2.1 of the License, or (at your option) any later version.
#
#    This library is distributed in the hope that it will be useful,
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
#    Lesser General Public License for more details.
#
#    You should have received a copy of the GNU Lesser General Public
#    License along with this library; if not, write to the Free Software
#    Foundation, Inc., 59 Temple Place, Suite 330, Boston,
#    MA 02111-1307  USA
#

import contextlib
import io
import os, unittest
import random
import shutil
import sqlite3
import string
import tempfile

from dlg import droputils
from dlg.ddap_protocol import DROPStates, ExecutionMode, AppDROPStates
from dlg.drop import FileDROP, AppDROP, InMemoryDROP, \
    NullDROP, BarrierAppDROP, \
    DirectoryContainer, ContainerDROP, InputFiredAppDROP, RDBMSDrop
from dlg.droputils import DROPWaiterCtx
from dlg.exceptions import InvalidDropException
from dlg.apps.simple import NullBarrierApp, SimpleBranch, SleepAndCopyApp


try:
    from crc32c import crc32
except:
    from binascii import crc32

ONE_MB = 1024 ** 2

def _start_ns_thread(ns_daemon):
    ns_daemon.requestLoop()

def isContainer(drop):
    # return isinstance(drop, ContainerDROP)
    # A Pyro-friendly way to check for a ContainerDROP is to see if
    # invoking its 'children' attribute fails or not
    try:
        drop.children
        return True
    except AttributeError:
        return False

class SumupContainerChecksum(BarrierAppDROP):
    """
    A dummy BarrierAppDROP that recursively sums up the checksums of
    all the individual DROPs it consumes, and then stores the final
    result in its output DROP
    """
    def run(self):
        crcSum = 0
        for inputDrop in self.inputs:
            if inputDrop.status == DROPStates.COMPLETED:
                crcSum += inputDrop.checksum
        outputDrop = self.outputs[0]
        outputDrop.write(str(crcSum).encode('utf8'))

class TestDROP(unittest.TestCase):

    def setUp(self):
        """
        library-specific setup
        """
        self._test_drop_sz = 16 # MB
        self._test_block_sz =  2 # MB
        self._test_num_blocks = self._test_drop_sz // self._test_block_sz
        self._test_block = os.urandom(self._test_block_sz * ONE_MB)

    def tearDown(self):
        shutil.rmtree("/tmp/daliuge_tfiles", True)

    def test_NullDROP(self):
        """
        Check that the NullDROP is usable for testing
        """
        a = NullDROP('A', 'A', expectedSize=5)
        a.write(b"1234")
        a.write(b"5")
        allContents = droputils.allDropContents(a)
        self.assertFalse(allContents)

    def test_write_FileDROP(self):
        """
        Test a FileDROP and a simple AppDROP (for checksum calculation)
        """
        self._test_write_withDropType(FileDROP)

    def test_write_InMemoryDROP(self):
        """
        Test an InMemoryDROP and a simple AppDROP (for checksum calculation)
        """
        self._test_write_withDropType(InMemoryDROP)

    def _test_write_withDropType(self, dropType):
        """
        Test an AbstractDROP and a simple AppDROP (for checksum calculation)
        """
        a = dropType('oid:A', 'uid:A', expectedSize = self._test_drop_sz * ONE_MB)
        b = SumupContainerChecksum('oid:B', 'uid:B')
        c = InMemoryDROP('oid:C', 'uid:C')
        b.addInput(a)
        b.addOutput(c)

        test_crc = 0
        with DROPWaiterCtx(self, c):
            for _ in range(self._test_num_blocks):
                a.write(self._test_block)
                test_crc = crc32(self._test_block, test_crc)

        # Read the checksum from c
        cChecksum = int(droputils.allDropContents(c))

        self.assertNotEqual(a.checksum, 0)
        self.assertEqual(a.checksum, test_crc)<commit_msg>add basic plasma drop tests<commit_after>#
#    ICRAR - International Centre for Radio Astronomy Research
#    (c) UWA - The University of Western Australia, 2015
#    Copyright by UWA (in the framework of the ICRAR)
#    All rights reserved
#
#    This library is free software; you can redistribute it and/or
#    modify it under the terms of the GNU Lesser General Public
#    License as published by the Free Software Foundation; either
#    version 2.1 of the License, or (at your option) any later version.
#
#    This library is distributed in the hope that it will be useful,
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
#    Lesser General Public License for more details.
#
#    You should have received a copy of the GNU Lesser General Public
#    License along with this library; if not, write to the Free Software
#    Foundation, Inc., 59 Temple Place, Suite 330, Boston,
#    MA 02111-1307  USA
#

import contextlib
import io
import os, unittest
import random
import shutil
import sqlite3
import string
import tempfile
import subprocess

from dlg import droputils
from dlg.ddap_protocol import DROPStates, ExecutionMode, AppDROPStates
from dlg.drop import FileDROP, AppDROP, InMemoryDROP, PlasmaDROP, \
    PlasmaFlightDROP, NullDROP, BarrierAppDROP, \
    DirectoryContainer, ContainerDROP, InputFiredAppDROP, RDBMSDrop
from dlg.droputils import DROPWaiterCtx
from dlg.exceptions import InvalidDropException
from dlg.apps.simple import NullBarrierApp, SimpleBranch, SleepAndCopyApp


try:
    from crc32c import crc32
except:
    from binascii import crc32

ONE_MB = 1024 ** 2

def _start_ns_thread(ns_daemon):
    ns_daemon.requestLoop()

def isContainer(drop):
    # return isinstance(drop, ContainerDROP)
    # A Pyro-friendly way to check for a ContainerDROP is to see if
    # invoking its 'children' attribute fails or not
    try:
        drop.children
        return True
    except AttributeError:
        return False

class SumupContainerChecksum(BarrierAppDROP):
    """
    A dummy BarrierAppDROP that recursively sums up the checksums of
    all the individual DROPs it consumes, and then stores the final
    result in its output DROP
    """
    def run(self):
        crcSum = 0
        for inputDrop in self.inputs:
            if inputDrop.status == DROPStates.COMPLETED:
                crcSum += inputDrop.checksum
        outputDrop = self.outputs[0]
        outputDrop.write(str(crcSum).encode('utf8'))

class TestDROP(unittest.TestCase):

    def setUp(self):
        """
        library-specific setup
        """
        self._test_drop_sz = 16 # MB
        self._test_block_sz =  2 # MB
        self._test_num_blocks = self._test_drop_sz // self._test_block_sz
        self._test_block = os.urandom(self._test_block_sz * ONE_MB)

    def tearDown(self):
        shutil.rmtree("/tmp/daliuge_tfiles", True)

    def test_NullDROP(self):
        """
        Check that the NullDROP is usable for testing
        """
        a = NullDROP('A', 'A', expectedSize=5)
        a.write(b"1234")
        a.write(b"5")
        allContents = droputils.allDropContents(a)
        self.assertFalse(allContents)

    def test_write_FileDROP(self):
        """
        Test a FileDROP and a simple AppDROP (for checksum calculation)
        """
        self._test_write_withDropType(FileDROP)

    def test_write_InMemoryDROP(self):
        """
        Test an InMemoryDROP and a simple AppDROP (for checksum calculation)
        """
        self._test_write_withDropType(InMemoryDROP)

    def test_write_plasmaDROP(self):
        """
        Test an PlasmaDrop and a simple AppDROP (for checksum calculation)
        """
        self._test_drop_sz = 16 # MB
        self._test_block_sz =  16 # MB
        self._test_num_blocks = self._test_drop_sz // self._test_block_sz
        self._test_block = os.urandom(self._test_block_sz * ONE_MB)
        try:
            store = subprocess.Popen(["plasma_store", "-m", "100000000", "-s", "/tmp/plasma"])
            self._test_write_withDropType(PlasmaDROP)
        finally:
            store.terminate()

    def test_write_plasmaFlightDROP(self):
        """
        Test an PlasmaDrop and a simple AppDROP (for checksum calculation)
        """
        self._test_drop_sz = 16 # MB
        self._test_block_sz =  16 # MB
        self._test_num_blocks = self._test_drop_sz // self._test_block_sz
        self._test_block = os.urandom(self._test_block_sz * ONE_MB)
        try:
            store = subprocess.Popen(["plasma_store", "-m", "100000000", "-s", "/tmp/plasma"])
            self._test_write_withDropType(PlasmaFlightDROP)
        finally:
            store.terminate()

    def _test_write_withDropType(self, dropType):
        """
        Test an AbstractDROP and a simple AppDROP (for checksum calculation)
        """
        a = dropType('oid:A', 'uid:A', expectedSize = self._test_drop_sz * ONE_MB)
        b = SumupContainerChecksum('oid:B', 'uid:B')
        c = InMemoryDROP('oid:C', 'uid:C')
        b.addInput(a)
        b.addOutput(c)

        test_crc = 0
        with DROPWaiterCtx(self, c):
            for _ in range(self._test_num_blocks):
                a.write(self._test_block)
                test_crc = crc32(self._test_block, test_crc)

        # Read the checksum from c
        cChecksum = int(droputils.allDropContents(c))

        self.assertNotEqual(a.checksum, 0)
        self.assertEqual(a.checksum, test_crc)
{'fraction_non_alphanumeric': 0.08038258038258038, 'fraction_numerical': 0.012413512413512413, 'mean_word_length': 2.7818391689111195, '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': '5915139', 'n_tokens_mistral': 3087, 'n_tokens_neox': 2974, 'n_words': 887}