''' Created on Jul 21, 2010 @author: sau ''' import unittest import os import sqlite3 as sqlite from erange import ReadDataset testDBName = "testRDS.rds" rnaTestDBName = "testRDSRNA.rds" class TestReadDataset(unittest.TestCase): def setUp(self): self.rds = ReadDataset.ReadDataset(testDBName, initialize=True, datasetType="DNA", verbose=False) self.rnaRds = ReadDataset.ReadDataset(rnaTestDBName, initialize=True, datasetType="RNA", verbose=False) def tearDown(self): del(self.rds) os.remove(testDBName) del(self.rnaRds) os.remove(rnaTestDBName) #TODO: rename and integrate def testZeeNewStuff(self): rdsEntryList = [("testRead", "chr1", 1, 100, "+", 1.0, "", ""), ("dup start", "chr1", 1, 150, "+", 1.0, "", ""), ("new read", "chr1", 80, 100, "+", 1.0, "", ""), ("testRead", "chr2", 201, 400, "+", 1.0, "", ""), ("dup start", "chr2", 201, 450, "+", 1.0, "", ""), ("new read", "chr2", 280, 400, "+", 1.0, "", ""), ("three up", "chr3", 1, 80, "+", 1.0, "", ""), ("three two", "chr3", 201, 230, "+", 1.0, "", "") ] self.rds.insertUniqs(rdsEntryList) dbcon = sqlite.connect(testDBName) sql = dbcon.cursor() sql.execute("select chrom,start from uniqs group by chrom,start having ( count(start) > 1 and count(chrom) > 1)") result = [("chr1", 1), ("chr2", 201)] for eachEntry in sql.fetchall(): self.assertTrue(eachEntry in result) sql.execute("select chrom,start from uniqs group by chrom,start having ( count(start) = 1 and count(chrom) = 1)") result = [("chr1", 80), ("chr2", 280), ("chr3", 1), ("chr3", 201)] for eachEntry in sql.fetchall(): self.assertTrue(eachEntry in result) sql.execute("select * from uniqs group by chrom,start having ( count(start) > 1 and count(chrom) > 1) union select * from uniqs group by chrom,start having ( count(start) = 1 and count(chrom) = 1)") result = [(2, "dup start", "chr1", 1, 150, "+", 1.0, "", ""), (3, "new read", "chr1", 80, 100, "+", 1.0, "", ""), (5, "dup start", "chr2", 201, 450, "+", 1.0, "", ""), (6, "new read", "chr2", 280, 400, "+", 1.0, "", ""), (7, "three up", "chr3", 1, 80, "+", 1.0, "", ""), (8, "three two", "chr3", 201, 230, "+", 1.0, "", "") ] for eachEntry in sql.fetchall(): self.assertTrue(eachEntry in result) sql.execute("select chrom,start from uniqs where start > 100 group by chrom,start having ( count(start) > 1 and count(chrom) > 1) order by chrom,start") result = [("chr2", 201)] for eachEntry in sql.fetchall(): self.assertTrue(eachEntry in result) rdsEntryList = [("testMultiRead", "chr1", 1, 200, "+", 0.5, "", ""), ("testMultiRead", "chr1", 1, 200, "+", 0.5, "", ""), ("testMultiRead", "chr2", 80, 200, "+", 0.5, "", ""), ("testMultiRead", "chr2", 1, 200, "+", 0.5, "", ""), ("testMultiRead", "chr2", 5000, 25000, "+", 0.5, "", ""), ("testMultiRead", "chr3", 1, 200, "+", 0.5, "", ""), ("testMultiRead", "chr3", 70, 500, "+", 0.5, "", "") ] self.rds.insertMulti(rdsEntryList) sql.execute("select chrom,start from (select chrom,start from uniqs union all select chrom,start from multi) group by chrom,start having ( count(start) > 1 and count(chrom) > 1)") result = [("chr1", 1), ("chr2", 201), ("chr3", 1)] for eachEntry in sql.fetchall(): self.assertTrue(eachEntry in result) sql.execute("select chrom,start from (select chrom,start from uniqs union all select chrom,start from multi) group by chrom,start having ( count(start) = 1 and count(chrom) = 1)") result = [("chr1", 80), ("chr2", 1), ("chr2", 80), ("chr2", 280), ("chr2", 5000), ("chr3", 70), ("chr3", 201) ] for eachEntry in sql.fetchall(): self.assertTrue(eachEntry in result) sql.execute("select chrom,start from (select chrom,start from uniqs union all select chrom,start from multi) group by chrom,start having ( count(start) > 1 and count(chrom) > 1) union select chrom,start from (select chrom,start from uniqs union all select chrom,start from multi) group by chrom,start having ( count(start) = 1 and count(chrom) = 1)") result = sql.fetchall() result = [("chr1", 1), ("chr1", 80), ("chr2", 1), ("chr2", 80), ("chr2", 201), ("chr2", 280), ("chr2", 5000), ("chr3", 1), ("chr3", 70), ("chr3", 201) ] for eachEntry in sql.fetchall(): self.assertTrue(eachEntry in result) result = {"1": [{"start": 1, "sense": "+"}, {"start": 80, "sense": "+"}], "3": [{"start": 1, "sense": "+"}, {"start": 70, "sense": "+"}, {"start": 201, "sense": "+"}], "2": [{"start": 1, "sense": "+"}, {"start": 80, "sense": "+"}, {"start": 201, "sense": "+"}, {"start": 280, "sense": "+"}, {"start": 5000, "sense": "+"}] } self.assertEquals(result, self.rds.getReadsDict(combine5p=True, doMulti=True)) print self.rds.getReadsDict(combine5p=True, doMulti=True, withWeight=True) def testReadDatasetBuiltIns(self): # Initialize an existing rds file self.assertRaises(sqlite.OperationalError, ReadDataset.ReadDataset, testDBName, initialize=True, datasetType="DNA", verbose=True) self.assertEquals(0, len(self.rds)) rdsEntryList = [("testRead", "chr1", 1, 100, "+", 1.0, "", "")] self.rds.insertUniqs(rdsEntryList) self.assertEquals(1, len(self.rds)) rdsEntryList = [("testMultiRead", "chr1", 101, 200, "+", 0.5, "", ""), ("testMultiRead", "chr1", 101, 200, "+", 0.5, "", "")] self.rds.insertMulti(rdsEntryList) self.assertEquals(2, len(self.rds)) rdsEntryList = [("testSpliceRead", "chr1", 1000, 1100, 1150, 1200, "+", 1.0, "", "")] self.assertRaises(sqlite.OperationalError, self.rds.insertSplices, rdsEntryList) self.rnaRds.insertSplices(rdsEntryList) self.assertEquals(2, len(self.rds)) self.assertEquals(1, len(self.rnaRds)) def testInsertUniqs(self): rdsEntryList = [("testRead", "chr1", 1, 100, "+", 1.0, "", "")] self.rds.insertUniqs(rdsEntryList) self.assertEquals(1, len(self.rds)) rdsEntryList = [("testRead2", "chr1", 200, 300, "+", 1.0, "", "")] self.rds.insertUniqs(rdsEntryList) self.assertEquals(2, len(self.rds)) def testInsertMulti(self): rdsEntryList = [("testMultiRead", "chr1", 101, 200, "+", 0.5, "", ""), ("testMultiRead", "chr1", 101, 200, "+", 0.5, "", "")] self.rds.insertMulti(rdsEntryList) self.assertEquals(1, len(self.rds)) def testInsertSplices(self): rdsEntryList = [("testSpliceRead", "chr1", 1000, 1100, 1150, 1200, "+", 1.0, "", "")] self.assertRaises(sqlite.OperationalError, self.rds.insertSplices, rdsEntryList) self.rnaRds.insertSplices(rdsEntryList) self.assertEquals(0, len(self.rds)) self.assertEquals(1, len(self.rnaRds)) def testGetChromosomes(self): result = [] self.assertEqual(result, self.rds.getChromosomes(table="uniqs", fullChrom=True)) rdsEntryList = [("testRead", "chr1", 1, 100, "+", 1.0, "", "")] self.rds.insertUniqs(rdsEntryList) result = ["chr1"] self.assertEqual(result, self.rds.getChromosomes(table="uniqs", fullChrom=True)) self.assertRaises(sqlite.OperationalError, self.rds.getChromosomes, table="badTableName") #TODO: write unit test def testAttachDB(self): pass #TODO: write unit test def testDetachDB(self): pass #TODO: write unit test def testImportFromDB(self): pass def testGetTables(self): result = ["metadata", "uniqs", "multi"] self.assertEquals(result, self.rds.getTables()) result = ["metadata", "uniqs", "multi", "splices"] self.assertEquals(result, self.rnaRds.getTables()) def testHasIndex(self): self.assertFalse(self.rds.hasIndex()) self.rds.buildIndex() self.assertTrue(self.rds.hasIndex()) def testGetMetadata(self): returnDict = self.rds.getMetadata() self.assertTrue(returnDict.has_key("rdsVersion")) self.assertEquals(returnDict["dataType"], "DNA") result = {"dataType": "RNA"} self.assertEquals(result, self.rnaRds.getMetadata("dataType")) result = {} self.assertEquals(result, self.rds.getMetadata("badMetaDataName")) def testGetReadSize(self): self.assertRaises(ReadDataset.ReadDatasetError, self.rds.getReadSize) self.rds.insertMetadata([("readsize", "100")]) self.assertEquals(100, self.rds.getReadSize()) self.rds.updateMetadata("readsize", 100) self.assertEquals(100, self.rds.getReadSize()) self.rds.updateMetadata("readsize", "100 import") self.assertEquals(100, self.rds.getReadSize()) self.rds.updateMetadata("readsize", "badReadSize") self.assertRaises(ValueError, self.rds.getReadSize) def testGetDefaultCacheSize(self): self.assertEquals(100000, self.rds.getDefaultCacheSize()) def testGetMaxCoordinate(self): self.assertEquals(0, self.rnaRds.getMaxCoordinate("chr1")) rdsEntryList = [("testRead", "chr1", 1, 100, "+", 1.0, "", "")] self.rnaRds.insertUniqs(rdsEntryList) self.assertEquals(1, self.rnaRds.getMaxCoordinate("chr1")) self.assertEquals(0, self.rnaRds.getMaxCoordinate("chr2")) self.assertEquals(0, self.rnaRds.getMaxCoordinate("chr1", doUniqs=False)) rdsEntryList = [("testMultiRead", "chr1", 101, 200, "+", 0.5, "", ""), ("testMultiRead", "chr1", 101, 200, "+", 0.5, "", "")] self.rnaRds.insertMulti(rdsEntryList) self.assertEquals(1, self.rnaRds.getMaxCoordinate("chr1")) self.assertEquals(101, self.rnaRds.getMaxCoordinate("chr1", doMulti=True)) rdsEntryList = [("testSpliceRead", "chr1", 1000, 1100, 1150, 1200, "+", 1.0, "", "")] self.rnaRds.insertSplices(rdsEntryList) self.assertEquals(1, self.rnaRds.getMaxCoordinate("chr1")) self.assertEquals(101, self.rnaRds.getMaxCoordinate("chr1", doMulti=True)) self.assertEquals(1150, self.rnaRds.getMaxCoordinate("chr1", doSplices=True)) def testGetReadsDict(self): self.assertEquals({}, self.rds.getReadsDict()) rdsEntryList = [("testRead", "chr1", 1, 100, "+", 1.0, "", "")] self.rds.insertUniqs(rdsEntryList) reads = self.rds.getReadsDict() self.assertEquals(1, len(reads)) self.assertEquals(1, len(reads["1"])) read = reads["1"][0] self.assertEquals(["start", "sense"], read.keys()) self.assertEquals(1, read["start"]) self.assertEquals("+", read["sense"]) reads = self.rds.getReadsDict(bothEnds=True, noSense=False, fullChrom=True, withWeight=True, withFlag=True, withMismatch=True, withID=True, withChrom=True, readIDDict=True) self.assertEquals(1, len(reads)) self.assertEquals(1, len(reads["testRead"])) read = reads["testRead"][0] self.assertEquals(["readID", "weight", "stop", "mismatch","start", "flag","sense", "chrom"], read.keys()) self.assertEquals("testRead", read["readID"]) self.assertEquals(1.0, read["weight"]) self.assertEquals(100, read["stop"]) self.assertEquals("", read["mismatch"]) self.assertEquals(1, read["start"]) self.assertEquals("", read["flag"]) self.assertEquals("+", read["sense"]) self.assertEquals("chr1", read["chrom"]) self.assertEquals({}, self.rds.getReadsDict(hasMismatch=True)) self.assertEquals({}, self.rds.getReadsDict(strand="-")) self.assertEquals(1, len(self.rds.getReadsDict(strand="+"))) rdsEntryList = [("testRead2", "chr1", 201, 300, "-", 1.0, "A", "G22A")] self.rds.insertUniqs(rdsEntryList) reads = self.rds.getReadsDict() self.assertEquals(1, len(reads)) reads = self.rds.getReadsDict() self.assertEquals(2, len(reads["1"])) read = reads["1"][1] self.assertEquals(201, read["start"]) reads = self.rds.getReadsDict(strand="+") self.assertEquals(1, len(reads)) read = reads["1"][0] self.assertEquals("+", read["sense"]) reads = self.rds.getReadsDict(strand="-") self.assertEquals(1, len(reads)) reads = self.rds.getReadsDict(start=199) self.assertEquals(1, len(reads["1"])) reads = self.rds.getReadsDict(hasMismatch=True) self.assertEquals(1, len(reads["1"])) rdsEntryList = [("testMultiRead", "chr2", 101, 200, "+", 0.5, "", ""), ("testMultiRead", "chr2", 101, 200, "+", 0.5, "", "")] self.rds.insertMulti(rdsEntryList) reads = self.rds.getReadsDict() self.assertEquals(1, len(reads)) reads = self.rds.getReadsDict(doMulti=True) self.assertEquals(2, len(reads)) reads = self.rds.getReadsDict(doUniqs=False, doMulti=True) self.assertFalse(reads.has_key("1")) def testGetSplicesDict(self): self.assertRaises(sqlite.OperationalError, self.rds.getSplicesDict) rdsEntryList = [("testSpliceRead", "chr1", 1000, 1100, 1150, 1200, "+", 1.0, "", "")] self.rnaRds.insertSplices(rdsEntryList) reads = self.rnaRds.getSplicesDict() self.assertEquals(1, len(reads)) self.assertEquals(1, len(reads["1"])) read = reads["1"][0] result = ["startR", "stopL", "sense", "startL", "stopR"] self.assertEquals(result, read.keys()) self.assertEquals(1000, read["startL"]) self.assertEquals("+", read["sense"]) reads = self.rnaRds.getSplicesDict(splitRead=True) self.assertEquals(2, len(reads["1"])) self.assertEquals(1000, reads["1"][0]["startL"]) self.assertFalse(reads["1"][0].has_key("startR")) self.assertFalse(reads["1"][0].has_key("stopR")) self.assertEquals(1150, reads["1"][1]["startR"]) self.assertFalse(reads["1"][1].has_key("startL")) self.assertFalse(reads["1"][1].has_key("stopL")) self.assertEquals(reads["1"][0]["sense"], reads["1"][1]["sense"]) reads = self.rnaRds.getSplicesDict(noSense=False, fullChrom=True, withWeight=True, withFlag=True, withMismatch=True, withID=True, withChrom=True, readIDDict=True) self.assertEquals(1, len(reads)) self.assertEquals(1, len(reads["testSpliceRead"])) read = reads["testSpliceRead"][0] result = ["readID", "weight", "startR", "mismatch","stopR", "stopL", "flag", "startL", "sense", "chrom"] self.assertEquals(result, read.keys()) self.assertEquals("testSpliceRead", read["readID"]) self.assertEquals(1.0, read["weight"]) self.assertEquals(1150, read["startR"]) self.assertEquals("", read["mismatch"]) self.assertEquals(1200, read["stopR"]) self.assertEquals(1100, read["stopL"]) self.assertEquals("", read["flag"]) self.assertEquals(1000, read["startL"]) self.assertEquals("+", read["sense"]) self.assertEquals("chr1", read["chrom"]) self.assertEquals({}, self.rnaRds.getSplicesDict(hasMismatch=True)) self.assertEquals({}, self.rnaRds.getSplicesDict(strand="-")) self.assertEquals(1, len(self.rnaRds.getSplicesDict(strand="+"))) rdsEntryList = [("testSpliceRead2", "chr1", 2000, 2100, 2150, 2200, "-", 1.0, "A", "G20T")] self.rnaRds.insertSplices(rdsEntryList) reads = self.rnaRds.getSplicesDict() self.assertEquals(1, len(reads)) reads = self.rnaRds.getSplicesDict() self.assertEquals(2, len(reads["1"])) read = reads["1"][1] self.assertEquals(2000, read["startL"]) reads = self.rnaRds.getSplicesDict(strand="+") self.assertEquals(1, len(reads)) read = reads["1"][0] self.assertEquals("+", read["sense"]) reads = self.rnaRds.getSplicesDict(strand="-") self.assertEquals(1, len(reads)) reads = self.rnaRds.getSplicesDict(start=1199) self.assertEquals(1, len(reads["1"])) reads = self.rnaRds.getSplicesDict(hasMismatch=True) self.assertEquals(1, len(reads["1"])) rdsEntryList = [("testSpliceRead3", "chr2", 2000, 2100, 2150, 2200, "-", 1.0, "A", "G20T")] self.rnaRds.insertSplices(rdsEntryList) reads = self.rnaRds.getSplicesDict() self.assertEquals(2, len(reads)) self.assertEquals(2, len(reads["1"])) self.assertEquals(1, len(reads["2"])) reads = self.rnaRds.getSplicesDict(withID=True, chrom="chr2") self.assertFalse(reads.has_key("1")) self.assertEquals("testSpliceRead3", reads["2"][0]["readID"]) def testGetCounts(self): self.assertEquals(0, self.rds.getCounts()) self.assertEquals((0, 0, 0), self.rds.getCounts(multi=True, reportCombined=False)) rdsEntryList = [("testRead", "chr1", 1, 100, "+", 1.0, "", "")] self.rds.insertUniqs(rdsEntryList) self.assertEquals(1, self.rds.getCounts()) self.assertEquals((1, 0, 0), self.rds.getCounts(multi=True, reportCombined=False)) rdsEntryList = [("testMultiRead", "chr1", 101, 200, "+", 0.5, "", ""), ("testMultiRead", "chr1", 101, 200, "+", 0.5, "", "")] self.rds.insertMulti(rdsEntryList) self.assertEquals(2, self.rds.getCounts(multi=True)) self.assertEquals((1, 1, 0), self.rds.getCounts(multi=True, reportCombined=False)) self.assertEquals(1, self.rds.getCounts(chrom="chr1")) self.assertEquals(0, self.rds.getCounts(chrom="chr2")) self.assertEquals(1, self.rds.getCounts(rmin=1)) self.assertEquals(1, self.rds.getCounts(rmin=1, rmax=1000)) self.assertEquals(1, self.rds.getCounts(rmax=1000)) self.assertEquals(0, self.rds.getCounts(rmin=1000)) self.assertEquals(0, self.rds.getCounts(rmax=0)) self.assertEquals(1, self.rds.getCounts(sense="+")) self.assertEquals(0, self.rds.getCounts(sense="-")) self.assertEquals(0, self.rnaRds.getCounts()) rdsEntryList = [("testSpliceRead", "chr1", 1000, 1100, 1150, 1200, "+", 1.0, "", "")] self.rnaRds.insertSplices(rdsEntryList) self.assertEquals(1, self.rnaRds.getCounts(splices=True)) def testGetTotalCounts(self): self.assertEquals(0, self.rds.getTotalCounts()) rdsEntryList = [("testRead", "chr1", 1, 100, "+", 1.0, "", "")] self.rds.insertUniqs(rdsEntryList) self.assertEquals(1, self.rds.getTotalCounts()) rdsEntryList = [("testMultiRead", "chr1", 101, 200, "+", 0.5, "", ""), ("testMultiRead", "chr1", 101, 200, "+", 0.5, "", "")] self.rds.insertMulti(rdsEntryList) self.assertEquals(2, self.rds.getTotalCounts()) self.assertEquals(2, self.rds.getTotalCounts(chrom="chr1")) self.assertEquals(0, self.rds.getTotalCounts(chrom="chr2")) self.assertEquals(2, self.rds.getTotalCounts(rmin=1)) self.assertEquals(2, self.rds.getTotalCounts(rmax=1000)) self.assertEquals(1, self.rds.getTotalCounts(rmin=101, rmax=1000)) self.assertEquals(1, self.rds.getTotalCounts(rmin=1, rmax=100)) self.assertEquals(0, self.rds.getTotalCounts(rmin=1000)) self.assertEquals(0, self.rds.getTotalCounts(rmax=0)) def testGetTableEntryCount(self): table = "uniqs" self.assertEquals(0, self.rds.getTableEntryCount(table)) rdsEntryList = [("testRead", "chr1", 1, 100, "+", 1.0, "", "")] self.rds.insertUniqs(rdsEntryList) self.assertEquals(1, self.rds.getTableEntryCount(table)) rdsEntryList = [("testMultiRead", "chr1", 101, 200, "+", 0.5, "", ""), ("testMultiRead", "chr1", 101, 200, "+", 0.5, "", "")] self.rds.insertMulti(rdsEntryList) self.assertEquals(1, self.rds.getTableEntryCount(table)) self.assertEquals(1, self.rds.getTableEntryCount(table, chrom="chr1")) self.assertEquals(0, self.rds.getTableEntryCount(table, chrom="chr2")) self.assertEquals(1, self.rds.getTableEntryCount(table, rmin=1)) self.assertEquals(1, self.rds.getTableEntryCount(table, rmax=1000)) self.assertEquals(0, self.rds.getTableEntryCount(table, rmin=101, rmax=1000)) self.assertEquals(0, self.rds.getTableEntryCount(table, rmin=1000)) self.assertEquals(0, self.rds.getTableEntryCount(table, rmax=0)) self.assertEquals(1, self.rds.getTableEntryCount(table, restrict=" sense ='+' ")) self.assertEquals(0, self.rds.getTableEntryCount(table, restrict=" sense ='-' ")) self.assertEquals(1, self.rds.getTableEntryCount(table, distinct=True)) table="multi" self.assertEquals(1, self.rds.getTableEntryCount(table)) self.assertEquals(1, self.rds.getTableEntryCount(table, chrom="chr1")) self.assertEquals(0, self.rds.getTableEntryCount(table, chrom="chr2")) self.assertEquals(1, self.rds.getTableEntryCount(table, rmin=1)) self.assertEquals(1, self.rds.getTableEntryCount(table, rmax=1000)) self.assertEquals(1, self.rds.getTableEntryCount(table, rmin=101, rmax=1000)) self.assertEquals(0, self.rds.getTableEntryCount(table, rmin=1000)) self.assertEquals(0, self.rds.getTableEntryCount(table, rmax=0)) self.assertEquals(1, self.rds.getTableEntryCount(table, restrict=" sense ='+' ")) self.assertEquals(0, self.rds.getTableEntryCount(table, restrict=" sense ='-' ")) self.assertEquals(1, self.rds.getTableEntryCount(table, distinct=True)) rdsEntryList = [("testSpliceRead", "chr1", 1000, 1100, 1150, 1200, "+", 1.0, "", "")] self.rnaRds.insertSplices(rdsEntryList) table="splices" self.assertEquals(1, self.rnaRds.getTableEntryCount(table)) self.assertEquals(1, self.rnaRds.getTableEntryCount(table, chrom="chr1")) self.assertEquals(0, self.rnaRds.getTableEntryCount(table, chrom="chr2")) self.assertEquals(1, self.rnaRds.getTableEntryCount(table, rmin=1, startField="startL")) self.assertRaises(sqlite.OperationalError, self.rnaRds.getTableEntryCount, table, rmin=1) self.assertEquals(1, self.rnaRds.getTableEntryCount(table, rmax=2000, startField="startL")) self.assertRaises(sqlite.OperationalError, self.rnaRds.getTableEntryCount, table, rmax=2000) self.assertEquals(0, self.rnaRds.getTableEntryCount(table, rmax=999, startField="startL")) self.assertEquals(1, self.rnaRds.getTableEntryCount(table, rmin=1000, startField="startL")) self.assertEquals(0, self.rnaRds.getTableEntryCount(table, rmax=0, startField="startL")) self.assertEquals(1, self.rnaRds.getTableEntryCount(table, restrict=" sense ='+' ")) self.assertEquals(0, self.rnaRds.getTableEntryCount(table, restrict=" sense ='-' ")) self.assertEquals(1, self.rnaRds.getTableEntryCount(table, distinct=True, startField="startL")) self.assertRaises(sqlite.OperationalError, self.rnaRds.getTableEntryCount, table, distinct=True) def testGetUniqsCount(self): self.assertEquals(0, self.rds.getUniqsCount()) rdsEntryList = [("testRead", "chr1", 1, 100, "+", 1.0, "", "")] self.rds.insertUniqs(rdsEntryList) self.assertEquals(1, self.rds.getUniqsCount()) rdsEntryList = [("testMultiRead", "chr1", 101, 200, "+", 0.5, "", ""), ("testMultiRead", "chr1", 101, 200, "+", 0.5, "", "")] self.rds.insertMulti(rdsEntryList) self.assertEquals(1, self.rds.getUniqsCount()) self.assertEquals(1, self.rds.getUniqsCount(chrom="chr1")) self.assertEquals(0, self.rds.getUniqsCount(chrom="chr2")) self.assertEquals(1, self.rds.getUniqsCount(rmin=1)) self.assertEquals(1, self.rds.getUniqsCount(rmax=1000)) self.assertEquals(0, self.rds.getUniqsCount(rmin=101, rmax=1000)) self.assertEquals(0, self.rds.getUniqsCount(rmin=1000)) self.assertEquals(0, self.rds.getUniqsCount(rmax=0)) self.assertEquals(1, self.rds.getUniqsCount(restrict=" sense ='+' ")) self.assertEquals(0, self.rds.getUniqsCount(restrict=" sense ='-' ")) self.assertEquals(1, self.rds.getUniqsCount(distinct=True)) def testGetSplicesCount(self): self.assertEquals(0, self.rnaRds.getSplicesCount()) rdsEntryList = [("testRead", "chr1", 1, 100, "+", 1.0, "", "")] self.rnaRds.insertUniqs(rdsEntryList) self.assertEquals(0, self.rnaRds.getSplicesCount()) rdsEntryList = [("testMultiRead", "chr1", 101, 200, "+", 0.5, "", ""), ("testMultiRead", "chr1", 101, 200, "+", 0.5, "", "")] self.rnaRds.insertMulti(rdsEntryList) self.assertEquals(0, self.rnaRds.getSplicesCount()) rdsEntryList = [("testSpliceRead", "chr1", 1000, 1100, 1150, 1200, "+", 1.0, "", "")] self.rnaRds.insertSplices(rdsEntryList) self.assertEquals(1, self.rnaRds.getSplicesCount()) self.assertEquals(1, self.rnaRds.getSplicesCount(chrom="chr1")) self.assertEquals(0, self.rnaRds.getSplicesCount(chrom="chr2")) self.assertEquals(1, self.rnaRds.getSplicesCount(rmin=1)) self.assertEquals(1, self.rnaRds.getSplicesCount(rmax=2000)) self.assertEquals(0, self.rnaRds.getSplicesCount(rmax=999)) self.assertEquals(1, self.rnaRds.getSplicesCount(rmin=1000)) self.assertEquals(0, self.rnaRds.getSplicesCount(rmax=0)) self.assertEquals(1, self.rnaRds.getSplicesCount(restrict=" sense ='+' ")) self.assertEquals(0, self.rnaRds.getSplicesCount(restrict=" sense ='-' ")) self.assertEquals(1, self.rnaRds.getSplicesCount(distinct=True)) def testGetMultiCount(self): self.assertEquals(0, self.rds.getMultiCount()) rdsEntryList = [("testRead", "chr1", 1, 100, "+", 1.0, "", "")] self.rds.insertUniqs(rdsEntryList) self.assertEquals(0, self.rds.getMultiCount()) rdsEntryList = [("testMultiRead", "chr1", 101, 200, "+", 0.5, "", ""), ("testMultiRead", "chr1", 101, 200, "+", 0.5, "", "")] self.rds.insertMulti(rdsEntryList) self.assertEquals(1, self.rds.getMultiCount()) self.assertEquals(1, self.rds.getMultiCount(chrom="chr1")) self.assertEquals(0, self.rds.getMultiCount(chrom="chr2")) self.assertEquals(1, self.rds.getMultiCount(rmin=1)) self.assertEquals(1, self.rds.getMultiCount(rmax=1000)) self.assertEquals(0, self.rds.getMultiCount(rmin=1, rmax=100)) self.assertEquals(0, self.rds.getMultiCount(rmin=1000)) self.assertEquals(0, self.rds.getMultiCount(rmax=0)) self.assertEquals(1, self.rds.getMultiCount(restrict=" sense ='+' ")) self.assertEquals(0, self.rds.getMultiCount(restrict=" sense ='-' ")) self.assertEquals(1, self.rds.getMultiCount(distinct=True)) def testGetReadIDs(self): self.assertEquals([], self.rnaRds.getReadIDs()) rdsEntryList = [("testRead", "chr1", 1, 100, "+", 1.0, "", "")] self.rnaRds.insertUniqs(rdsEntryList) result = ["testRead"] self.assertEquals(result, self.rnaRds.getReadIDs()) rdsEntryList = [("testMultiRead", "chr1", 101, 200, "+", 0.5, "", ""), ("testMultiRead", "chr1", 101, 200, "+", 0.5, "", "")] self.rnaRds.insertMulti(rdsEntryList) result = ["testRead"] self.assertEquals(result, self.rnaRds.getReadIDs()) result = ["testMultiRead", "testRead"] self.assertEquals(result, self.rnaRds.getReadIDs(multi=True)) rdsEntryList = [("testRead2", "chr1", 201, 300, "+", 1.0, "", "")] self.rnaRds.insertUniqs(rdsEntryList) result = ["testRead", "testRead2"] self.assertEquals(result, self.rnaRds.getReadIDs()) result = ["testRead"] self.assertEquals(result, self.rnaRds.getReadIDs(limit=1)) result = ["testMultiRead"] self.assertEquals(result, self.rnaRds.getReadIDs(multi=True, limit=1)) rdsEntryList = [("testPair/1", "chr1", 301, 400, "+", 1.0, "", "")] self.rnaRds.insertUniqs(rdsEntryList) result = ["testPair", "testRead", "testRead2"] self.assertEquals(result, self.rnaRds.getReadIDs(paired=True)) rdsEntryList = [("testSpliceRead", "chr1", 1000, 1100, 1150, 1200, "+", 1.0, "", "")] self.rnaRds.insertSplices(rdsEntryList) result = ["testSpliceRead"] self.assertEquals(result, self.rnaRds.getReadIDs(uniqs=False, splices=True)) result = ["testPair/1", "testRead", "testRead2", "testSpliceRead"] self.assertEquals(result, self.rnaRds.getReadIDs(splices=True)) def testGetMismatches(self): self.assertRaises(ReadDataset.ReadDatasetError, self.rds.getMismatches) self.rds.insertMetadata([("readsize", "5")]) rdsEntryList = [("testRead", "chr1", 1, 5, "+", 1.0, "", "")] self.rds.insertUniqs(rdsEntryList) result = {"chr1": []} self.assertEquals(result, self.rds.getMismatches()) rdsEntryList = [("testRead", "chr1", 1, 5, "+", 1.0, "", "C3T")] self.rds.insertUniqs(rdsEntryList) result = {"chr1": [[1, 3, "T", "C"]]} self.assertEquals(result, self.rds.getMismatches()) result = {"chr2": []} self.assertEquals(result, self.rds.getMismatches(mischrom="chr2")) rdsEntryList = [("testRead", "chr1", 10, 15, "+", 1.0, "", "C3T")] self.rds.insertUniqs(rdsEntryList) result = {"chr1": [[1, 3, "T", "C"], [10, 12, "T", "C"]]} self.assertEquals(result, self.rds.getMismatches()) rdsEntryList = [("testRead", "chr2", 10, 15, "+", 1.0, "", "C3T")] self.rds.insertUniqs(rdsEntryList) result = {"chr1": [[1, 3, "T", "C"], [10, 12, "T", "C"]], "chr2": [[10, 12, "T", "C"]]} self.assertEquals(result, self.rds.getMismatches()) rdsEntryList = [("testSpliceRead", "chr1", 1000, 1100, 1150, 1200, "+", 1.0, "", "C41T")] self.rnaRds.insertSplices(rdsEntryList) self.rnaRds.insertMetadata([("readsize", "150")]) result = {"chr1": [[1000, 1040, "T", "C"]]} #TODO: This test case fails. If there are only splice entries for a chromosome it shouldn't # be necessary to specify the chromosome. #self.assertEquals(result, self.rnaRds.getMismatches()) self.assertEquals(result, self.rnaRds.getMismatches(mischrom="chr1")) #TODO: needs fixing up def testGetChromProfile(self): chromProfile = self.rds.getChromProfile("chr1") result = [] self.assertEquals(result, chromProfile.tolist()) rdsEntryList = [("testRead", "chr1", 1, 5, "+", 1.0, "", "")] self.rds.insertUniqs(rdsEntryList) chromProfile = self.rds.getChromProfile("chr1") result = [] self.assertEquals(result, chromProfile.tolist()) self.rds.insertMetadata([("readsize", "5")]) chromProfile = self.rds.getChromProfile("chr1") result = [0.0, 1.0, 1.0, 1.0, 1.0] self.assertEquals(result, chromProfile.tolist()) rdsEntryList = [("testRead2", "chr1", 7, 11, "+", 1.0, "", "")] self.rds.insertUniqs(rdsEntryList) # This doesn't seem to make sense the default behavior is to only get the first readlen bases chromProfile = self.rds.getChromProfile("chr1") result = [0.0, 1.0, 1.0, 1.0, 1.0] self.assertEquals(result, chromProfile.tolist()) # as it stands this doesn't see right either. Getting an indexError at currentpos 5. chromProfile = self.rds.getChromProfile("chr1", cstop=11) result = [0.0, 1.0, 1.0, 1.0, 1.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0] self.assertEquals(result, chromProfile.tolist()) def testInsertMetadata(self): result = {} self.assertEquals(result, self.rds.getMetadata("testMeta")) self.rds.insertMetadata([("testMeta", "100")]) result = {"testMeta": "100"} self.assertEquals(result, self.rds.getMetadata("testMeta")) self.rds.insertMetadata([("testMeta", "200")]) result = {"testMeta:2": "200", "testMeta": "100"} self.assertEquals(result, self.rds.getMetadata("testMeta")) def testUpdateMetadata(self): result = {} self.assertEquals(result, self.rds.getMetadata("testMeta")) self.rds.insertMetadata([("testMeta", "100")]) result = {"testMeta": "100"} self.assertEquals(result, self.rds.getMetadata("testMeta")) self.rds.updateMetadata("testMeta", "200") result = {"testMeta": "200"} self.assertEquals(result, self.rds.getMetadata("testMeta")) self.rds.updateMetadata("testMeta", "300", "200") result = {"testMeta": "300"} self.assertEquals(result, self.rds.getMetadata("testMeta")) self.rds.updateMetadata("testMeta", "200", "200") result = {"testMeta": "300"} self.assertEquals(result, self.rds.getMetadata("testMeta")) def testFlagReads(self): readData = self.rnaRds.getReadsDict(withFlag=True) self.assertEquals({}, readData) rdsEntryList = [("testRead", "chr1", 1, 100, "+", 1.0, "", "")] self.rnaRds.insertUniqs(rdsEntryList) result = [""] flags = self.getRDSFlags("1", self.rnaRds) self.assertEquals(result, flags) regions = [()] self.assertRaises(sqlite.ProgrammingError, self.rnaRds.flagReads, regions) regions = [("test", "chr1", "0", "1000")] self.rnaRds.flagReads(regions) result = ["test"] flags = self.getRDSFlags("1", self.rnaRds) self.assertEquals(result, flags) regions = [("test2", "chr1", "600", "1000")] self.rnaRds.flagReads(regions) result = ["test"] flags = self.getRDSFlags("1", self.rnaRds) self.assertEquals(result, flags) rdsEntryList = [("testRead2", "chr1", 101, 200, "+", 1.0, "", "")] self.rnaRds.insertUniqs(rdsEntryList) regions = [("test2", "chr1", "101", "1000")] self.rnaRds.flagReads(regions) result = ["test", "test2"] flags = self.getRDSFlags("1", self.rnaRds) self.assertEquals(result, flags) rdsEntryList = [("testMultiRead", "chr1", 201, 300, "+", 0.5, "", ""), ("testMultiRead", "chr1", 201, 300, "+", 0.5, "", "")] self.rnaRds.insertMulti(rdsEntryList) regions = [("test", "chr1", "0", "1000")] self.rnaRds.flagReads(regions) result = ["test", "test", "", ""] flags = self.getRDSFlags("1", self.rnaRds, doMulti=True) self.assertEquals(result, flags) regions = [("multi", "chr1", "1", "1000")] self.rnaRds.flagReads(regions, uniqs=False, multi=True) result = ["test", "test", "multi", "multi"] flags = self.getRDSFlags("1", self.rnaRds, doMulti=True) self.assertEquals(result, flags) rdsEntryList = [("testSpliceRead", "chr1", 1000, 1100, 1150, 1200, "+", 1.0, "", "")] self.rnaRds.insertSplices(rdsEntryList) regions = [("test", "chr1", "0", "1500")] self.rnaRds.flagReads(regions) result = ["", "test", "test", "multi", "multi"] flags = self.getRDSFlags("1", self.rnaRds, doMulti=True, splice=True) self.assertEquals(result, flags) regions = [("splice", "chr1", "1", "1500")] self.rnaRds.flagReads(regions, uniqs=False, multi=False, splices=True) result = [" L:splice R:splice", "test", "test", "multi", "multi"] flags = self.getRDSFlags("1", self.rnaRds, doMulti=True, splice=True) self.assertEquals(result, flags) rdsEntryList = [("testNegSense", "chr1", 301, 400, "-", 1.0, "", "")] self.rnaRds.insertUniqs(rdsEntryList) regions = [("test", "chr1", "0", "1500", "+")] self.rnaRds.flagReads(regions, sense="anythingBut'Both'") result = ["test", "test", ""] flags = self.getRDSFlags("1", self.rnaRds) self.assertEquals(result, flags) regions = [("neg", "chr1", "0", "1500", "-")] self.rnaRds.flagReads(regions, sense="anythingBut'Both'") result = ["test", "test", "neg"] flags = self.getRDSFlags("1", self.rnaRds) self.assertEquals(result, flags) def getRDSFlags(self, chromosome, rds, doMulti=False, splice=False): if splice: readData = rds.getSplicesDict(withFlag=True) else: readData = rds.getReadsDict(withFlag=True, doMulti=doMulti) flags = [] for read in readData[chromosome]: flags.append(read["flag"]) if splice: nonSplice = self.getRDSFlags(chromosome, rds, doMulti, splice=False) for flag in nonSplice: flags.append(flag) return flags def testSetFlags(self): rdsEntryList = [("test", "chr1", 1, 100, "+", 1.0, "uniq", "")] self.rds.insertUniqs(rdsEntryList) self.rnaRds.insertUniqs(rdsEntryList) rdsEntryList = [("testMultiRead", "chr1", 201, 300, "+", 0.5, "multi", ""), ("testMultiRead", "chr1", 201, 300, "+", 0.5, "multi", "")] self.rnaRds.insertMulti(rdsEntryList) rdsEntryList = [("testSpliceRead", "chr1", 1000, 1100, 1150, 1200, "+", 1.0, "splice", "")] self.rnaRds.insertSplices(rdsEntryList) result = ["reset"] self.rds.setFlags("reset") flags = self.getRDSFlags("1", self.rds) self.assertEquals(result, flags) result = ["splice", "uniq", "resetMulti", "resetMulti"] self.rnaRds.setFlags("resetMulti", uniqs=False, splices=False) flags = self.getRDSFlags("1", self.rnaRds, doMulti=True, splice=True) self.assertEquals(result, flags) result = ["resetAll", "resetAll", "resetAll", "resetAll"] self.rnaRds.setFlags("resetAll") flags = self.getRDSFlags("1", self.rnaRds, doMulti=True, splice=True) self.assertEquals(result, flags) def testResetFlags(self): rdsEntryList = [("test", "chr1", 1, 100, "+", 1.0, "uniq", "")] self.rds.insertUniqs(rdsEntryList) self.rnaRds.insertUniqs(rdsEntryList) rdsEntryList = [("testMultiRead", "chr1", 201, 300, "+", 0.5, "multi", ""), ("testMultiRead", "chr1", 201, 300, "+", 0.5, "multi", "")] self.rnaRds.insertMulti(rdsEntryList) rdsEntryList = [("testSpliceRead", "chr1", 1000, 1100, 1150, 1200, "+", 1.0, "splice", "")] self.rnaRds.insertSplices(rdsEntryList) self.rds.resetFlags() result = [""] flags = self.getRDSFlags("1", self.rds) self.assertEquals(result, flags) self.rnaRds.resetFlags() result = ["", "", ""] flags = self.getRDSFlags("1", self.rnaRds, doMulti=True) self.assertEquals(result, flags) self.rnaRds.resetFlags() result = ["", ""] flags = self.getRDSFlags("1", self.rnaRds, splice=True) self.assertEquals(result, flags) def testReweighMultireads(self): rdsEntryList = [("testMultiRead", "chr1", 201, 300, "+", 0.5, "multi", ""), ("testMultiRead", "chr1", 201, 300, "+", 0.5, "multi", "")] self.rds.insertMulti(rdsEntryList) readData = ("0.25", "chr1", "201", "testMultiRead") self.rds.reweighMultireads([readData]) readDict = self.rds.getReadsDict(withWeight=True, doMulti=True) read = readDict["1"][0] self.assertEquals(0.25, read["weight"]) #TODO: write unit test def testSetSynchronousPragma(self): pass #TODO: write unit test def testSetDBcache(self): pass #TODO: write unit test def testExecute(self): pass #TODO: write unit test def testExecuteCommit(self): pass def testBuildIndex(self): self.assertFalse(self.rds.hasIndex()) self.rds.buildIndex() self.assertTrue(self.rds.hasIndex()) def testDropIndex(self): self.assertFalse(self.rds.hasIndex()) self.rds.buildIndex() self.assertTrue(self.rds.hasIndex()) self.rds.dropIndex() self.assertFalse(self.rds.hasIndex()) self.assertFalse(self.rnaRds.hasIndex()) self.rnaRds.buildIndex() self.assertTrue(self.rnaRds.hasIndex()) self.rnaRds.dropIndex() self.assertFalse(self.rnaRds.hasIndex()) #TODO: write unit test def testMemSync(self): pass #TODO: write unit test def testCopyDBEntriesToMemory(self): pass #TODO: write unit test def testCopySpliceDBEntriesToMemory(self): pass def suite(): suite = unittest.TestSuite() suite.addTest(unittest.makeSuite(TestReadDataset)) return suite if __name__ == "__main__": #import sys;sys.argv = ['', 'Test.testName'] unittest.main()