def test_partition(self):
     a = Babe().pull(string=self.s, format='csv')
     a = a.partition(field='date')
     d = {}
     a.push(stream_dict=d, format="csv")
     self.assertEquals(d['2012-04-04'].getvalue(), 'date,name,value\n2012-04-04,John,1\n2012-04-04,Luke,2\n')
     self.assertEquals(d['2012-04-05'].getvalue(), 'date,name,value\n2012-04-05,John,1\n')

 def test_http(self):
     a = Babe().pull(protocol='http',
                     host='localhost',
                     name='Test',
                     filename='remote/files/test.csv',
                     port=self.port)
     self.assertEquals(a.to_string(), 'foo,bar,f,d\n1,2,3.2,2010/10/02\n3,4,1.2,2011/02/02\n')

 def test_user_agent(self):
     a = Babe().pull(string=self.s, format="csv")
     a = a.user_agent(field="useragent",
                      output_os="os",
                      output_browser="browser",
                      output_browser_version="browser_version")
     self.assertEquals(a.to_string(), self.s2)

 def test_join_none(self):
     a = Babe().pull(string=self.s1, format='csv')
     a = a.join(join_stream=Babe().pull(string=self.s2_bis, format='csv'),
                key='country',
                join_key='country_code',
                on_error=Babe.ON_ERROR_NONE)
     self.assertEquals(a.to_string(), self.sjoined_bis)

 def test_twitter(self):
     a = Babe().pull_twitter()
     a = a.filterColumns(keep_fields=
     ["author_name", "author_id", "author_screen_name", "created_at", "hashtags", "text", "in_reply_to_status_id_str"])
     a = a.typedetect()
     buf = StringIO()
     a.push(stream=buf, format='csv')

 def test_replace(self):
     a = Babe().pull(filename='tests/test.csv', name='Test').typedetect()
     a = a.mapTo(lambda row : [row.foo+1, row.bar*2], fields=['a','b'])
     buf = StringIO()
     a.push(stream=buf, format='csv')
     s = """a,b\n2,4\n4,8\n"""
     self.assertEquals(buf.getvalue(), s)

    def test_html(self):
        a = Babe().pull(string=self.s, format="csv")
        self.assertEqual(a.to_string(format="html"), """<h2></h2><table>
<tr><th>a</th><th>b</th></tr>
<tr><td>1</td><td>2</td></tr>
</table>
""")

 def test_bulk(self):
     a = Babe().pull(stream=StringIO(self.s), format="csv")
     a = a.typedetect()
     a = a.bulkMapTo(lambda list: [[sum([r.a for r in list])]] * len(list),
                     bulk_size=2,
                     insert_fields=["b"])
     self.assertEquals(a.to_string(), self.s2)

 def test_multi2(self):
     a = Babe()
     a = a.pull(stream=StringIO(self.s), format='csv').pull(string=self.s, format='csv')
     a = a.merge_substreams()
     buf = StringIO()
     a.push(stream=buf, format='csv')
     self.assertEquals(buf.getvalue(), self.s2)

 def test_zip(self):
     babe = Babe()
     a = babe.pull(stream=StringIO(self.s), format="csv")
     a.push(filename='tests/test.zip')
     b = Babe().pull(filename='tests/test.zip')
     buf = StringIO()
     b.push(stream=buf)
     self.assertEquals(buf.getvalue(), self.s)

 def test_load(self):
     start_time = '2012-04-23 11:00'
     end_time = '2012-04-23 12:00'
     a = Babe().pull_kontagent(start_time, end_time, sample_mode=True)
     buf = StringIO()
     a = a.head(n=10)
     a.push(stream=buf, format='csv')
     print buf.getvalue()

 def test_buzzdata(self):
     a = Babe().pull(protocol='buzzdata', 
             dataroom='best-city-contest-worldwide-cost-of-living-index',
             uuid='aINAPyLGur4y37yAyCM7w3', 
              username='eiu', format='xls')
     a = a.head(2)
     buf = StringIO()
     a.push(stream=buf, format='csv')

    def test_insert(self):
        a = Babe().pull(filename='tests/test.csv', name='Test').typedetect()
        a = a.mapTo(lambda row : row.foo+1, insert_fields=['fooplus'])
        s = """foo,bar,f,d,fooplus
1,2,3.2,2010/10/02,2
3,4,1.2,2011/02/02,4
"""
        self.assertEquals(a.to_string(), s)

 def test_sort(self):
     babe = Babe()
     s = '\n'.join(['k,v'] + [ '%u,%u' % (i,-i) for i in xrange(0,10001)])
     a = babe.pull(string=s, name='test', format='csv')
     a = a.typedetect()
     a = a.sort(field='v')
     a = a.head(n=1)
     self.assertEquals(a.to_string(), 'k,v\n10000,-10000\n')

 def test_sort(self):
     babe = Babe()
     s = "\n".join(["k,v"] + ["%u,%u" % (i, -i) for i in xrange(0, 10001)])
     a = babe.pull(string=s, name="test", format="csv")
     a = a.typedetect()
     a = a.sort(field="v")
     a = a.head(n=1)
     self.assertEquals(a.to_string(), "k,v\n10000,-10000\n")

    def test_tuple(self):
        a = Babe().pull(filename='tests/test.csv', name='Test').typedetect()
        a = a.mapTo(lambda obj: obj._replace(foo=obj.foo + 1))
        s = """foo,bar,f,d
2,2,3.2,2010/10/02
4,4,1.2,2011/02/02
"""
        self.assertEquals(a.to_string(),  s)

 def test_ftp(self):
     babe = Babe()
     a = babe.pull(filename='tests/test.csv', name='Test')
     a.push(filename='test.csv', protocol='ftp', user=self.user, password=self.password, host='localhost', port=self.port, protocol_early_check= False)
     b = babe.pull(filename='test.csv', name='Test', protocol='ftp', user=self.user, password=self.password, host='localhost', port=self.port)
     buf = StringIO()
     b.push(stream=buf, format='csv')
     self.assertEquals(buf.getvalue(), test_csv_content)

 def test_load_partition(self):
     start_time = "2012-04-23 11:00"
     end_time = "2012-04-23 12:00"
     a = Babe().pull_kontagent(start_time, end_time, sample_mode=True)
     a = a.head(n=10)
     d = {}
     a.push(stream_dict=d, format="csv")
     self.assertEquals(list(d.keys()), ["2012-04-23_11"])

 def test_s3(self):
     s = "a,b\n1,2\n3,4\n"
     a = Babe().pull(string=s,
                     format='csv',
                     name='Test')
     a.push(filename='test_gs.csv',
            bucket='bertrandtest',
            delimiter="\t",
            protocol="gs")

 def test_parse(self):
     a = Babe().pull(stream=StringIO(self.s), format='csv')
     a = a.parse_time(field="time",
                      output_time="time",
                      output_date="date",
                      output_hour="hour",
                      input_timezone="CET",
                      output_timezone="GMT")
     self.assertEquals(a.to_string(), self.s2)